WIP

parent 71acea27
...@@ -99,11 +99,11 @@ class ConnectionHelper: ...@@ -99,11 +99,11 @@ class ConnectionHelper:
body += f.read() body += f.read()
f.close() f.close()
body += '\r\n' body += '\r\n'
self.connection.request("POST", self.conn.path + '/' + path, self.connection.request("POST", self.conn.path + '/' + path,
body, header_dict) body, header_dict)
self.response = self.connection.getresponse() self.response = self.connection.getresponse()
class ERP5TestReportHandler: class ERP5TestReportHandler:
def __init__(self, url, suite_name): def __init__(self, url, suite_name):
# random test id # random test id
...@@ -113,6 +113,7 @@ class ERP5TestReportHandler: ...@@ -113,6 +113,7 @@ class ERP5TestReportHandler:
) )
self.connection_helper = ConnectionHelper(url) self.connection_helper = ConnectionHelper(url)
self.suite_name = suite_name self.suite_name = suite_name
self.test_id = "20111026-18C84076DE29E8"
def reportStart(self): def reportStart(self):
# report that test is running # report that test is running
...@@ -124,27 +125,62 @@ class ERP5TestReportHandler: ...@@ -124,27 +125,62 @@ class ERP5TestReportHandler:
def reportFinished(self, out_file): def reportFinished(self, out_file):
# make file parsable by erp5_test_results # make file parsable by erp5_test_results
detail, success, failure, error_title_list = self.processResult(out_file) out_file, success, failure, duration, error_title_list = self.processResult(
log = """Title:%s out_file)
Success: %d
Failure: %d #tempout = tempfile.mkstemp()[1]
Info: %s""" % (error_title_list, success, failure, detail) #templog = tempfile.mkstemp()[1]
templog = tempfile.mkstemp()[1] #log_lines = open(out_file, 'r').readlines()
tl = open(templog, 'w') #tl = open(templog, 'w')
tl.write(log) #tl.write(TB_SEP + '\n')
tl.close() #for log_line in log_lines:
# starts = log_line.startswith
# if starts('Ran') or starts('FAILED') or starts('OK') or starts(TB_SEP):
# continue
# if starts('ERROR: ') or starts('FAIL: '):
# tl.write('internal-test: ' + log_line)
# continue
# tl.write(log_line)
#
#tl.write("----------------------------------------------------------------------\n")
#tl.write('Ran %s test in %.2fs\n' % (test_count, duration))
#if success:
# tl.write('OK\n')
#else:
# tl.write('FAILED (failures=%s)\n' % failure)
#tl.write(TB_SEP + '\n')
# create nice zip archive # create nice zip archive
tempzip = tempfile.mkstemp()[1] tempzip = tempfile.mkstemp()[1]
zip = zipfile.ZipFile(tempzip, 'w') zip = zipfile.ZipFile(tempzip, 'w')
zip.write(templog, '%s/002/stderr' % self.suite_name)
# temperr is dummy
temperr = tempfile.mkstemp()[1]
open(temperr, 'w')
i = 1
test_case_list = [dict(name="first", log="OK\n"), dict(name="second", log="FAILED (failures=1)\n")]
for test_case in test_case_list:
tempcmd = tempfile.mkstemp()[1]
tempout = tempfile.mkstemp()[1]
open(tempcmd, 'w').write(test_case['name'])
open(tempout, 'w').write(test_case['log'])
#XXX-Cedric : support cases for more than 9 test cases
zip.write(tempcmd, '%s/00%s/cmdline' % (self.suite_name, i))
zip.write(tempout, '%s/00%s/stderr' % (self.suite_name, i))
zip.write(temperr, '%s/00%s/stdout' % (self.suite_name, i))
os.unlink(tempcmd)
os.unlink(tempout)
i = i + 1
zip.close() zip.close()
os.unlink(templog) os.unlink(temperr)
# post it to ERP5 # post it to ERP5
self.connection_helper.POST('TestResultModule_reportCompleted', dict( self.connection_helper.POST('TestResultModule_reportCompleted', dict(
test_report_id=self.test_id), test_report_id=self.test_id),
file_list=[('filepath', tempzip)] file_list=[('filepath', tempzip)]
) )
import pdb; pdb.set_trace()
os.unlink(tempzip) os.unlink(tempzip)
def processResult(self, out_file): def processResult(self, out_file):
...@@ -153,6 +189,8 @@ Info: %s""" % (error_title_list, success, failure, detail) ...@@ -153,6 +189,8 @@ Info: %s""" % (error_title_list, success, failure, detail)
failure_amount = TEST_FAILURE_RE.search(file_content).group(1) failure_amount = TEST_FAILURE_RE.search(file_content).group(1)
error_title_list = [re.compile('\s+').sub(' ', x).strip() error_title_list = [re.compile('\s+').sub(' ', x).strip()
for x in TEST_ERROR_TITLE_RE.findall(file_content)] for x in TEST_ERROR_TITLE_RE.findall(file_content)]
# XXX-Cedric : duration
duration = 0
detail = '' detail = ''
for test_result in TEST_RESULT_RE.findall(file_content): for test_result in TEST_RESULT_RE.findall(file_content):
...@@ -169,4 +207,5 @@ Info: %s""" % (error_title_list, success, failure, detail) ...@@ -169,4 +207,5 @@ Info: %s""" % (error_title_list, success, failure, detail)
<body>%s</body> <body>%s</body>
</html>''' % detail </html>''' % detail
return detail, int(sucess_amount), int(failure_amount), error_title_list return detail, int(sucess_amount), int(failure_amount), duration, \
error_title_list
...@@ -38,17 +38,18 @@ def run(args): ...@@ -38,17 +38,18 @@ def run(args):
config['user'], config['password']) config['user'], config['password'])
while True: while True:
erp5_report = ERP5TestReportHandler(test_url, config['suite_name']) erp5_report = ERP5TestReportHandler(config['test_suite_master_url'],
import pdb; pdb.set_trace() config['project'] + '@' + config['suite_name'])
# Clean old test results # Clean old test results
openUrl('%s/TestTool_cleanUpTestResults' % (config['base_url'])) openUrl('%s/TestTool_cleanUpTestResults?__ac_name=%s&__ac_password=%s' % (
config['base_url'], config['user'], config['password']))
# TODO assert getresult is None # TODO assert getresult is None
#XXX-Cedric : clean firefox data (so that it does not load previous session) #XXX-Cedric : clean firefox data (so that it does not load previous session)
xvfb = Popen([config['xvfb_binary'], config['display']], stdout=PIPE) #xvfb = Popen([config['xvfb_binary'], config['display']], stdout=PIPE)
os.environ['DISPLAY'] = config['display'] #os.environ['DISPLAY'] = config['display']
sleep(10) #sleep(10)
command = [] command = []
command.append(config['browser_binary']) command.append(config['browser_binary'])
...@@ -61,11 +62,16 @@ def run(args): ...@@ -61,11 +62,16 @@ def run(args):
# Wait for test to be finished # Wait for test to be finished
while getStatus(config['base_url']) is '': while getStatus(config['base_url']) is '':
sleep(10) sleep(1) #10
import pdb; pdb.set_trace()
erp5_report.reportFinished(getStatus(config['base_url']).encode("utf-8", erp5_report.reportFinished(getStatus(config['base_url']).encode("utf-8",
"replace")) "replace"))
browser.kill() browser.kill()
terminateXvfb(xvfb, config['display']) #terminateXvfb(xvfb, config['display'])
print("Test finished and report sent, sleeping.")
sleep(3600) sleep(3600)
def openUrl(url): def openUrl(url):
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment