123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292 |
- #
- # Copyright (C) 2016 Intel Corporation
- #
- # SPDX-License-Identifier: MIT
- #
- import os
- import time
- import unittest
- import logging
- import re
- import json
- import sys
- from unittest import TextTestResult as _TestResult
- from unittest import TextTestRunner as _TestRunner
- class OEStreamLogger(object):
- def __init__(self, logger):
- self.logger = logger
- self.buffer = ""
- def write(self, msg):
- if len(msg) > 1 and msg[0] != '\n':
- if '...' in msg:
- self.buffer += msg
- elif self.buffer:
- self.buffer += msg
- self.logger.log(logging.INFO, self.buffer)
- self.buffer = ""
- else:
- self.logger.log(logging.INFO, msg)
- def flush(self):
- for handler in self.logger.handlers:
- handler.flush()
- class OETestResult(_TestResult):
- def __init__(self, tc, *args, **kwargs):
- super(OETestResult, self).__init__(*args, **kwargs)
- self.successes = []
- self.starttime = {}
- self.endtime = {}
- self.progressinfo = {}
- # Inject into tc so that TestDepends decorator can see results
- tc.results = self
- self.tc = tc
- # stdout and stderr for each test case
- self.logged_output = {}
- def startTest(self, test):
- # May have been set by concurrencytest
- if test.id() not in self.starttime:
- self.starttime[test.id()] = time.time()
- super(OETestResult, self).startTest(test)
- def stopTest(self, test):
- self.endtime[test.id()] = time.time()
- if self.buffer:
- self.logged_output[test.id()] = (
- sys.stdout.getvalue(), sys.stderr.getvalue())
- super(OETestResult, self).stopTest(test)
- if test.id() in self.progressinfo:
- self.tc.logger.info(self.progressinfo[test.id()])
- # Print the errors/failures early to aid/speed debugging, its a pain
- # to wait until selftest finishes to see them.
- for t in ['failures', 'errors', 'skipped', 'expectedFailures']:
- for (scase, msg) in getattr(self, t):
- if test.id() == scase.id():
- self.tc.logger.info(str(msg))
- break
- def logSummary(self, component, context_msg=''):
- elapsed_time = self.tc._run_end_time - self.tc._run_start_time
- self.tc.logger.info("SUMMARY:")
- self.tc.logger.info("%s (%s) - Ran %d test%s in %.3fs" % (component,
- context_msg, self.testsRun, self.testsRun != 1 and "s" or "",
- elapsed_time))
- if self.wasSuccessful():
- msg = "%s - OK - All required tests passed" % component
- else:
- msg = "%s - FAIL - Required tests failed" % component
- msg += " (successes=%d, skipped=%d, failures=%d, errors=%d)" % (len(self.successes), len(self.skipped), len(self.failures), len(self.errors))
- self.tc.logger.info(msg)
- def _getTestResultDetails(self, case):
- result_types = {'failures': 'FAILED', 'errors': 'ERROR', 'skipped': 'SKIPPED',
- 'expectedFailures': 'EXPECTEDFAIL', 'successes': 'PASSED'}
- for rtype in result_types:
- found = False
- for (scase, msg) in getattr(self, rtype):
- if case.id() == scase.id():
- found = True
- break
- scase_str = str(scase.id())
- # When fails at module or class level the class name is passed as string
- # so figure out to see if match
- m = re.search(r"^setUpModule \((?P<module_name>.*)\)$", scase_str)
- if m:
- if case.__class__.__module__ == m.group('module_name'):
- found = True
- break
- m = re.search(r"^setUpClass \((?P<class_name>.*)\)$", scase_str)
- if m:
- class_name = "%s.%s" % (case.__class__.__module__,
- case.__class__.__name__)
- if class_name == m.group('class_name'):
- found = True
- break
- if found:
- return result_types[rtype], msg
- return 'UNKNOWN', None
- def addSuccess(self, test):
- #Added so we can keep track of successes too
- self.successes.append((test, None))
- super(OETestResult, self).addSuccess(test)
- def logDetails(self, json_file_dir=None, configuration=None, result_id=None,
- dump_streams=False):
- self.tc.logger.info("RESULTS:")
- result = {}
- logs = {}
- if hasattr(self.tc, "extraresults"):
- result = self.tc.extraresults
- for case_name in self.tc._registry['cases']:
- case = self.tc._registry['cases'][case_name]
- (status, log) = self._getTestResultDetails(case)
- t = ""
- if case.id() in self.starttime and case.id() in self.endtime:
- t = " (" + "{0:.2f}".format(self.endtime[case.id()] - self.starttime[case.id()]) + "s)"
- if status not in logs:
- logs[status] = []
- logs[status].append("RESULTS - %s: %s%s" % (case.id(), status, t))
- report = {'status': status}
- if log:
- report['log'] = log
- if dump_streams and case.id() in self.logged_output:
- (stdout, stderr) = self.logged_output[case.id()]
- report['stdout'] = stdout
- report['stderr'] = stderr
- result[case.id()] = report
- for i in ['PASSED', 'SKIPPED', 'EXPECTEDFAIL', 'ERROR', 'FAILED', 'UNKNOWN']:
- if i not in logs:
- continue
- for l in logs[i]:
- self.tc.logger.info(l)
- if json_file_dir:
- tresultjsonhelper = OETestResultJSONHelper()
- tresultjsonhelper.dump_testresult_file(json_file_dir, configuration, result_id, result)
- def wasSuccessful(self):
- # Override as we unexpected successes aren't failures for us
- return (len(self.failures) == len(self.errors) == 0)
- class OEListTestsResult(object):
- def wasSuccessful(self):
- return True
- class OETestRunner(_TestRunner):
- streamLoggerClass = OEStreamLogger
- def __init__(self, tc, *args, **kwargs):
- kwargs['stream'] = self.streamLoggerClass(tc.logger)
- super(OETestRunner, self).__init__(*args, **kwargs)
- self.tc = tc
- self.resultclass = OETestResult
- def _makeResult(self):
- return self.resultclass(self.tc, self.stream, self.descriptions,
- self.verbosity)
- def _walk_suite(self, suite, func):
- for obj in suite:
- if isinstance(obj, unittest.suite.TestSuite):
- if len(obj._tests):
- self._walk_suite(obj, func)
- elif isinstance(obj, unittest.case.TestCase):
- func(self.tc.logger, obj)
- self._walked_cases = self._walked_cases + 1
- def _list_tests_name(self, suite):
- from oeqa.core.decorator.oetag import OETestTag
- self._walked_cases = 0
- def _list_cases(logger, case):
- oetag = None
- if hasattr(case, 'decorators'):
- for d in case.decorators:
- if isinstance(d, OETestTag):
- oetag = d.oetag
- logger.info("%s\t\t%s" % (oetag, case.id()))
- self.tc.logger.info("Listing all available tests:")
- self._walked_cases = 0
- self.tc.logger.info("id\ttag\t\ttest")
- self.tc.logger.info("-" * 80)
- self._walk_suite(suite, _list_cases)
- self.tc.logger.info("-" * 80)
- self.tc.logger.info("Total found:\t%s" % self._walked_cases)
- def _list_tests_class(self, suite):
- self._walked_cases = 0
- curr = {}
- def _list_classes(logger, case):
- if not 'module' in curr or curr['module'] != case.__module__:
- curr['module'] = case.__module__
- logger.info(curr['module'])
- if not 'class' in curr or curr['class'] != \
- case.__class__.__name__:
- curr['class'] = case.__class__.__name__
- logger.info(" -- %s" % curr['class'])
- logger.info(" -- -- %s" % case._testMethodName)
- self.tc.logger.info("Listing all available test classes:")
- self._walk_suite(suite, _list_classes)
- def _list_tests_module(self, suite):
- self._walked_cases = 0
- listed = []
- def _list_modules(logger, case):
- if not case.__module__ in listed:
- if case.__module__.startswith('_'):
- logger.info("%s (hidden)" % case.__module__)
- else:
- logger.info(case.__module__)
- listed.append(case.__module__)
- self.tc.logger.info("Listing all available test modules:")
- self._walk_suite(suite, _list_modules)
- def list_tests(self, suite, display_type):
- if display_type == 'name':
- self._list_tests_name(suite)
- elif display_type == 'class':
- self._list_tests_class(suite)
- elif display_type == 'module':
- self._list_tests_module(suite)
- return OEListTestsResult()
- class OETestResultJSONHelper(object):
- testresult_filename = 'testresults.json'
- def _get_existing_testresults_if_available(self, write_dir):
- testresults = {}
- file = os.path.join(write_dir, self.testresult_filename)
- if os.path.exists(file):
- with open(file, "r") as f:
- testresults = json.load(f)
- return testresults
- def _write_file(self, write_dir, file_name, file_content):
- file_path = os.path.join(write_dir, file_name)
- with open(file_path, 'w') as the_file:
- the_file.write(file_content)
- def dump_testresult_file(self, write_dir, configuration, result_id, test_result):
- bb.utils.mkdirhier(write_dir)
- lf = bb.utils.lockfile(os.path.join(write_dir, 'jsontestresult.lock'))
- test_results = self._get_existing_testresults_if_available(write_dir)
- test_results[result_id] = {'configuration': configuration, 'result': test_result}
- json_testresults = json.dumps(test_results, sort_keys=True, indent=4)
- self._write_file(write_dir, self.testresult_filename, json_testresults)
- bb.utils.unlockfile(lf)
|