From 50966fdf890219edaf06ae502abcb64f6d6e0351 Mon Sep 17 00:00:00 2001 From: Przemek Wirkus Date: Wed, 15 Oct 2014 16:05:57 +0100 Subject: [PATCH] Refactoring of report exporter to support HTML and JUNIT: Fixed issues with non ASCII characters read from serial port in ProcessObserver. Also changed reporting for errors and failures in TestSuite JUnit class --- workspace_tools/test_api.py | 52 +++++++++++++++++++++---------- workspace_tools/test_exporters.py | 13 +++++--- 2 files changed, 43 insertions(+), 22 deletions(-) diff --git a/workspace_tools/test_api.py b/workspace_tools/test_api.py index 52b667c55b..2f5bcd7ea5 100644 --- a/workspace_tools/test_api.py +++ b/workspace_tools/test_api.py @@ -873,6 +873,34 @@ class SingleTestRunner(object): Function also is pooling for serial port activity from process to catch all data printed by test runner and host test during test execution """ + + def get_char_from_queue(obs): + """ Get character from queue safe way + """ + try: + c = obs.queue.get(block=True, timeout=0.5) + except Empty, _: + c = None + return c + + def filter_queue_char(c): + """ Filters out non ASCII characters from serial port + """ + if ord(c) not in range(128): + c = ' ' + return c + + def get_test_result(output): + """ Parse test 'output' data + """ + result = self.TEST_RESULT_TIMEOUT + for line in "".join(output).splitlines(): + search_result = self.RE_DETECT_TESTCASE_RESULT.search(line) + if search_result and len(search_result.groups()): + result = self.TEST_RESULT_MAPPING[search_result.groups(0)[0]] + break + return result + # print "{%s} port:%s disk:%s" % (name, port, disk), cmd = ["python", "%s.py" % name, '-p', port, '-d', disk, '-t', str(duration)] @@ -896,15 +924,13 @@ class SingleTestRunner(object): line = '' output = [] while (time() - start_time) < duration: - try: - c = obs.queue.get(block=True, timeout=0.5) - except Empty, _: - c = None + c = get_char_from_queue(obs) if c: - output.append(c) if verbose: sys.stdout.write(c) + c = filter_queue_char(c) + output.append(c) # Give the mbed under test a way to communicate the end of the test if c in ['\n', '\r']: if '{end}' in line: @@ -913,28 +939,20 @@ class SingleTestRunner(object): else: line += c - try: - c = obs.queue.get(block=True, timeout=0.5) - except Empty, _: - c = None + c = get_char_from_queue(obs) if c: - output.append(c) if verbose: sys.stdout.write(c) + c = filter_queue_char(c) + output.append(c) if verbose: print "Test::Output::Finish" # Stop test process obs.stop() - # Parse test 'output' data - result = self.TEST_RESULT_TIMEOUT - for line in "".join(output).splitlines(): - search_result = self.RE_DETECT_TESTCASE_RESULT.search(line) - if search_result and len(search_result.groups()): - result = self.TEST_RESULT_MAPPING[search_result.groups(0)[0]] - break + result = get_test_result(output) return result, "".join(output) def is_peripherals_available(self, target_mcu_name, peripherals=None): diff --git a/workspace_tools/test_exporters.py b/workspace_tools/test_exporters.py index 04718d94c9..b83deca675 100644 --- a/workspace_tools/test_exporters.py +++ b/workspace_tools/test_exporters.py @@ -104,6 +104,7 @@ class ReportExporter(): tooltip_name = self.get_tooltip_name(test['toolchain_name'], test['target_name'], test['test_id'], test_no) background_color = RESULT_COLORS[test['single_test_result'] if test['single_test_result'] in RESULT_COLORS else 'ERROR'] result_div_style = "background-color: %s"% background_color + result = """
%s
@@ -121,7 +122,7 @@ class ReportExporter(): tooltip_name, test['test_description'], test['elapsed_time'], - unicode(test['single_test_output'], errors='ignore').replace('\n', '
')) + test['single_test_output'].replace('\n', '
')) return result def get_result_tree(self, test_results): @@ -216,17 +217,19 @@ class ReportExporter(): name = test_result['test_description'] classname = 'test.%s.%s.%s'% (target, toolchain, test_result['test_id']) elapsed_sec = test_result['elapsed_time'] - _stdout = unicode(test_result['single_test_output'], errors='ignore') + _stdout = test_result['single_test_output'] _stderr = '' tc = TestCase(name, classname, elapsed_sec, _stdout, _stderr) # Add extra failure / error info to test case result if test_result['single_test_result'] == 'FAIL': - tc.add_failure_info('Test failed', 'Test result: %s'% test_result['single_test_result']) + message = test_result['single_test_result'] + tc.add_failure_info(message, _stdout) elif test_result['single_test_result'] != 'OK': - tc.add_error_info('Test error', 'Test result: %s'% test_result['single_test_result']) + message = test_result['single_test_result'] + tc.add_error_info(message, _stdout) test_cases.append(tc) ts = TestSuite("test.suite.%s.%s"% (target, toolchain), test_cases, properties=test_suite_properties[target][toolchain]) test_suites.append(ts) - return TestSuite.to_xml_string(test_suites, encoding='utf8') + return TestSuite.to_xml_string(test_suites)