1# 2# Copyright (C) 2016 Intel Corporation 3# 4# SPDX-License-Identifier: MIT 5# 6 7import os 8import time 9import unittest 10import logging 11import re 12import json 13import sys 14 15from unittest import TextTestResult as _TestResult 16from unittest import TextTestRunner as _TestRunner 17 18class OEStreamLogger(object): 19 def __init__(self, logger): 20 self.logger = logger 21 self.buffer = "" 22 23 def write(self, msg): 24 if len(msg) > 1 and msg[0] != '\n': 25 if '...' in msg: 26 self.buffer += msg 27 elif self.buffer: 28 self.buffer += msg 29 self.logger.log(logging.INFO, self.buffer) 30 self.buffer = "" 31 else: 32 self.logger.log(logging.INFO, msg) 33 34 def flush(self): 35 for handler in self.logger.handlers: 36 handler.flush() 37 38class OETestResult(_TestResult): 39 def __init__(self, tc, *args, **kwargs): 40 super(OETestResult, self).__init__(*args, **kwargs) 41 42 self.successes = [] 43 self.starttime = {} 44 self.endtime = {} 45 self.progressinfo = {} 46 self.extraresults = {} 47 48 # Inject into tc so that TestDepends decorator can see results 49 tc.results = self 50 51 self.tc = tc 52 53 # stdout and stderr for each test case 54 self.logged_output = {} 55 56 def startTest(self, test): 57 # May have been set by concurrencytest 58 if test.id() not in self.starttime: 59 self.starttime[test.id()] = time.time() 60 super(OETestResult, self).startTest(test) 61 62 def stopTest(self, test): 63 self.endtime[test.id()] = time.time() 64 if self.buffer: 65 self.logged_output[test.id()] = ( 66 sys.stdout.getvalue(), sys.stderr.getvalue()) 67 super(OETestResult, self).stopTest(test) 68 if test.id() in self.progressinfo: 69 self.tc.logger.info(self.progressinfo[test.id()]) 70 71 # Print the errors/failures early to aid/speed debugging, its a pain 72 # to wait until selftest finishes to see them. 73 for t in ['failures', 'errors', 'skipped', 'expectedFailures']: 74 for (scase, msg) in getattr(self, t): 75 if test.id() == scase.id(): 76 self.tc.logger.info(str(msg)) 77 break 78 79 def logSummary(self, component, context_msg=''): 80 elapsed_time = self.tc._run_end_time - self.tc._run_start_time 81 self.tc.logger.info("SUMMARY:") 82 self.tc.logger.info("%s (%s) - Ran %d test%s in %.3fs" % (component, 83 context_msg, self.testsRun, self.testsRun != 1 and "s" or "", 84 elapsed_time)) 85 86 if self.wasSuccessful(): 87 msg = "%s - OK - All required tests passed" % component 88 else: 89 msg = "%s - FAIL - Required tests failed" % component 90 msg += " (successes=%d, skipped=%d, failures=%d, errors=%d)" % (len(self.successes), len(self.skipped), len(self.failures), len(self.errors)) 91 self.tc.logger.info(msg) 92 93 def _getTestResultDetails(self, case): 94 result_types = {'failures': 'FAILED', 'errors': 'ERROR', 'skipped': 'SKIPPED', 95 'expectedFailures': 'EXPECTEDFAIL', 'successes': 'PASSED', 96 'unexpectedSuccesses' : 'PASSED'} 97 98 for rtype in result_types: 99 found = False 100 for resultclass in getattr(self, rtype): 101 # unexpectedSuccesses are just lists, not lists of tuples 102 if isinstance(resultclass, tuple): 103 scase, msg = resultclass 104 else: 105 scase, msg = resultclass, None 106 if case.id() == scase.id(): 107 found = True 108 break 109 scase_str = str(scase.id()) 110 111 # When fails at module or class level the class name is passed as string 112 # so figure out to see if match 113 m = re.search(r"^setUpModule \((?P<module_name>.*)\).*$", scase_str) 114 if m: 115 if case.__class__.__module__ == m.group('module_name'): 116 found = True 117 break 118 119 m = re.search(r"^setUpClass \((?P<class_name>.*)\).*$", scase_str) 120 if m: 121 class_name = "%s.%s" % (case.__class__.__module__, 122 case.__class__.__name__) 123 124 if class_name == m.group('class_name'): 125 found = True 126 break 127 128 if found: 129 return result_types[rtype], msg 130 131 return 'UNKNOWN', None 132 133 def extractExtraResults(self, test, details = None): 134 extraresults = None 135 if details is not None and "extraresults" in details: 136 extraresults = details.get("extraresults", {}) 137 elif hasattr(test, "extraresults"): 138 extraresults = test.extraresults 139 140 if extraresults is not None: 141 for k, v in extraresults.items(): 142 # handle updating already existing entries (e.g. ptestresults.sections) 143 if k in self.extraresults: 144 self.extraresults[k].update(v) 145 else: 146 self.extraresults[k] = v 147 148 def addError(self, test, *args, details = None): 149 self.extractExtraResults(test, details = details) 150 return super(OETestResult, self).addError(test, *args) 151 152 def addFailure(self, test, *args, details = None): 153 self.extractExtraResults(test, details = details) 154 return super(OETestResult, self).addFailure(test, *args) 155 156 def addSuccess(self, test, details = None): 157 #Added so we can keep track of successes too 158 self.successes.append((test, None)) 159 self.extractExtraResults(test, details = details) 160 return super(OETestResult, self).addSuccess(test) 161 162 def addExpectedFailure(self, test, *args, details = None): 163 self.extractExtraResults(test, details = details) 164 return super(OETestResult, self).addExpectedFailure(test, *args) 165 166 def addUnexpectedSuccess(self, test, details = None): 167 self.extractExtraResults(test, details = details) 168 return super(OETestResult, self).addUnexpectedSuccess(test) 169 170 def logDetails(self, json_file_dir=None, configuration=None, result_id=None, 171 dump_streams=False): 172 self.tc.logger.info("RESULTS:") 173 174 result = self.extraresults 175 logs = {} 176 if hasattr(self.tc, "extraresults"): 177 result.update(self.tc.extraresults) 178 179 for case_name in self.tc._registry['cases']: 180 case = self.tc._registry['cases'][case_name] 181 182 (status, log) = self._getTestResultDetails(case) 183 184 t = "" 185 duration = 0 186 if case.id() in self.starttime and case.id() in self.endtime: 187 duration = self.endtime[case.id()] - self.starttime[case.id()] 188 t = " (" + "{0:.2f}".format(duration) + "s)" 189 190 if status not in logs: 191 logs[status] = [] 192 logs[status].append("RESULTS - %s: %s%s" % (case.id(), status, t)) 193 report = {'status': status} 194 if log: 195 report['log'] = log 196 if duration: 197 report['duration'] = duration 198 199 alltags = [] 200 # pull tags from the case class 201 if hasattr(case, "__oeqa_testtags"): 202 alltags.extend(getattr(case, "__oeqa_testtags")) 203 # pull tags from the method itself 204 test_name = case._testMethodName 205 if hasattr(case, test_name): 206 method = getattr(case, test_name) 207 if hasattr(method, "__oeqa_testtags"): 208 alltags.extend(getattr(method, "__oeqa_testtags")) 209 if alltags: 210 report['oetags'] = alltags 211 212 if dump_streams and case.id() in self.logged_output: 213 (stdout, stderr) = self.logged_output[case.id()] 214 report['stdout'] = stdout 215 report['stderr'] = stderr 216 result[case.id()] = report 217 218 for i in ['PASSED', 'SKIPPED', 'EXPECTEDFAIL', 'ERROR', 'FAILED', 'UNKNOWN']: 219 if i not in logs: 220 continue 221 for l in logs[i]: 222 self.tc.logger.info(l) 223 224 if json_file_dir: 225 tresultjsonhelper = OETestResultJSONHelper() 226 tresultjsonhelper.dump_testresult_file(json_file_dir, configuration, result_id, result) 227 228 def wasSuccessful(self): 229 # Override as we unexpected successes aren't failures for us 230 return (len(self.failures) == len(self.errors) == 0) 231 232class OEListTestsResult(object): 233 def wasSuccessful(self): 234 return True 235 236class OETestRunner(_TestRunner): 237 streamLoggerClass = OEStreamLogger 238 239 def __init__(self, tc, *args, **kwargs): 240 kwargs['stream'] = self.streamLoggerClass(tc.logger) 241 super(OETestRunner, self).__init__(*args, **kwargs) 242 self.tc = tc 243 self.resultclass = OETestResult 244 245 def _makeResult(self): 246 return self.resultclass(self.tc, self.stream, self.descriptions, 247 self.verbosity) 248 249 def _walk_suite(self, suite, func): 250 for obj in suite: 251 if isinstance(obj, unittest.suite.TestSuite): 252 if len(obj._tests): 253 self._walk_suite(obj, func) 254 elif isinstance(obj, unittest.case.TestCase): 255 func(self.tc.logger, obj) 256 self._walked_cases = self._walked_cases + 1 257 258 def _list_tests_name(self, suite): 259 self._walked_cases = 0 260 261 def _list_cases(logger, case): 262 oetags = [] 263 if hasattr(case, '__oeqa_testtags'): 264 oetags = getattr(case, '__oeqa_testtags') 265 if oetags: 266 logger.info("%s (%s)" % (case.id(), ",".join(oetags))) 267 else: 268 logger.info("%s" % (case.id())) 269 270 self.tc.logger.info("Listing all available tests:") 271 self._walked_cases = 0 272 self.tc.logger.info("test (tags)") 273 self.tc.logger.info("-" * 80) 274 self._walk_suite(suite, _list_cases) 275 self.tc.logger.info("-" * 80) 276 self.tc.logger.info("Total found:\t%s" % self._walked_cases) 277 278 def _list_tests_class(self, suite): 279 self._walked_cases = 0 280 281 curr = {} 282 def _list_classes(logger, case): 283 if not 'module' in curr or curr['module'] != case.__module__: 284 curr['module'] = case.__module__ 285 logger.info(curr['module']) 286 287 if not 'class' in curr or curr['class'] != \ 288 case.__class__.__name__: 289 curr['class'] = case.__class__.__name__ 290 logger.info(" -- %s" % curr['class']) 291 292 logger.info(" -- -- %s" % case._testMethodName) 293 294 self.tc.logger.info("Listing all available test classes:") 295 self._walk_suite(suite, _list_classes) 296 297 def _list_tests_module(self, suite): 298 self._walked_cases = 0 299 300 listed = [] 301 def _list_modules(logger, case): 302 if not case.__module__ in listed: 303 if case.__module__.startswith('_'): 304 logger.info("%s (hidden)" % case.__module__) 305 else: 306 logger.info(case.__module__) 307 listed.append(case.__module__) 308 309 self.tc.logger.info("Listing all available test modules:") 310 self._walk_suite(suite, _list_modules) 311 312 def list_tests(self, suite, display_type): 313 if display_type == 'name': 314 self._list_tests_name(suite) 315 elif display_type == 'class': 316 self._list_tests_class(suite) 317 elif display_type == 'module': 318 self._list_tests_module(suite) 319 320 return OEListTestsResult() 321 322class OETestResultJSONHelper(object): 323 324 testresult_filename = 'testresults.json' 325 326 def _get_existing_testresults_if_available(self, write_dir): 327 testresults = {} 328 file = os.path.join(write_dir, self.testresult_filename) 329 if os.path.exists(file): 330 with open(file, "r") as f: 331 testresults = json.load(f) 332 return testresults 333 334 def _write_file(self, write_dir, file_name, file_content): 335 file_path = os.path.join(write_dir, file_name) 336 with open(file_path, 'w') as the_file: 337 the_file.write(file_content) 338 339 def dump_testresult_file(self, write_dir, configuration, result_id, test_result): 340 try: 341 import bb 342 has_bb = True 343 bb.utils.mkdirhier(write_dir) 344 lf = bb.utils.lockfile(os.path.join(write_dir, 'jsontestresult.lock')) 345 except ImportError: 346 has_bb = False 347 os.makedirs(write_dir, exist_ok=True) 348 test_results = self._get_existing_testresults_if_available(write_dir) 349 test_results[result_id] = {'configuration': configuration, 'result': test_result} 350 json_testresults = json.dumps(test_results, sort_keys=True, indent=4) 351 self._write_file(write_dir, self.testresult_filename, json_testresults) 352 if has_bb: 353 bb.utils.unlockfile(lf) 354