diff options
Diffstat (limited to 'testing/mozbase/moztest/moztest/adapters/unit.py')
| -rw-r--r-- | testing/mozbase/moztest/moztest/adapters/unit.py | 217 |
1 files changed, 217 insertions, 0 deletions
diff --git a/testing/mozbase/moztest/moztest/adapters/unit.py b/testing/mozbase/moztest/moztest/adapters/unit.py new file mode 100644 index 0000000000..72c2f30052 --- /dev/null +++ b/testing/mozbase/moztest/moztest/adapters/unit.py @@ -0,0 +1,217 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +import sys +import time +import traceback +import unittest +from unittest import TextTestResult + +"""Adapter used to output structuredlog messages from unittest +testsuites""" + + +def get_test_class_name(test): + """ + This method is used to return the full class name from a + :class:`unittest.TestCase` instance. + + It is used as a default to define the "class_name" extra value + passed in structured loggers. You can override the default by + implementing a "get_test_class_name" method on you TestCase subclass. + """ + return "%s.%s" % (test.__class__.__module__, test.__class__.__name__) + + +def get_test_method_name(test): + """ + This method is used to return the full method name from a + :class:`unittest.TestCase` instance. + + It is used as a default to define the "method_name" extra value + passed in structured loggers. You can override the default by + implementing a "get_test_method_name" method on you TestCase subclass. + """ + return test._testMethodName + + +class StructuredTestResult(TextTestResult): + def __init__(self, *args, **kwargs): + self.logger = kwargs.pop("logger") + self.test_list = kwargs.pop("test_list", []) + self.result_callbacks = kwargs.pop("result_callbacks", []) + self.passed = 0 + self.testsRun = 0 + TextTestResult.__init__(self, *args, **kwargs) + + def call_callbacks(self, test, status): + debug_info = {} + for callback in self.result_callbacks: + info = callback(test, status) + if info is not None: + debug_info.update(info) + return debug_info + + def startTestRun(self): + # This would be an opportunity to call the logger's suite_start action, + # however some users may use multiple suites, and per the structured + # logging protocol, this action should only be called once. + pass + + def startTest(self, test): + self.testsRun += 1 + self.logger.test_start(test.id()) + + def stopTest(self, test): + pass + + def stopTestRun(self): + # This would be an opportunity to call the logger's suite_end action, + # however some users may use multiple suites, and per the structured + # logging protocol, this action should only be called once. + pass + + def _extract_err_message(self, err): + # Format an exception message in the style of unittest's _exc_info_to_string + # while maintaining a division between a traceback and a message. + exc_ty, val, _ = err + exc_msg = "".join(traceback.format_exception_only(exc_ty, val)) + if self.buffer: + output_msg = "\n".join([sys.stdout.getvalue(), sys.stderr.getvalue()]) + return "".join([exc_msg, output_msg]) + return exc_msg.rstrip() + + def _extract_stacktrace(self, err, test): + # Format an exception stack in the style of unittest's _exc_info_to_string + # while maintaining a division between a traceback and a message. + # This is mostly borrowed from unittest.result._exc_info_to_string. + + exctype, value, tb = err + while tb and self._is_relevant_tb_level(tb): + tb = tb.tb_next + # Header usually included by print_exception + lines = ["Traceback (most recent call last):\n"] + if exctype is test.failureException and hasattr( + self, "_count_relevant_tb_levels" + ): + length = self._count_relevant_tb_levels(tb) + lines += traceback.format_tb(tb, length) + else: + lines += traceback.format_tb(tb) + return "".join(lines) + + def _get_class_method_name(self, test): + if hasattr(test, "get_test_class_name"): + class_name = test.get_test_class_name() + else: + class_name = get_test_class_name(test) + + if hasattr(test, "get_test_method_name"): + method_name = test.get_test_method_name() + else: + method_name = get_test_method_name(test) + + return {"class_name": class_name, "method_name": method_name} + + def addError(self, test, err): + self.errors.append((test, self._exc_info_to_string(err, test))) + extra = self.call_callbacks(test, "ERROR") + extra.update(self._get_class_method_name(test)) + self.logger.test_end( + test.id(), + "ERROR", + message=self._extract_err_message(err), + expected="PASS", + stack=self._extract_stacktrace(err, test), + extra=extra, + ) + + def addFailure(self, test, err): + extra = self.call_callbacks(test, "FAIL") + extra.update(self._get_class_method_name(test)) + self.logger.test_end( + test.id(), + "FAIL", + message=self._extract_err_message(err), + expected="PASS", + stack=self._extract_stacktrace(err, test), + extra=extra, + ) + + def addSuccess(self, test): + extra = self._get_class_method_name(test) + self.logger.test_end(test.id(), "PASS", expected="PASS", extra=extra) + + def addExpectedFailure(self, test, err): + extra = self.call_callbacks(test, "FAIL") + extra.update(self._get_class_method_name(test)) + self.logger.test_end( + test.id(), + "FAIL", + message=self._extract_err_message(err), + expected="FAIL", + stack=self._extract_stacktrace(err, test), + extra=extra, + ) + + def addUnexpectedSuccess(self, test): + extra = self.call_callbacks(test, "PASS") + extra.update(self._get_class_method_name(test)) + self.logger.test_end(test.id(), "PASS", expected="FAIL", extra=extra) + + def addSkip(self, test, reason): + extra = self.call_callbacks(test, "SKIP") + extra.update(self._get_class_method_name(test)) + self.logger.test_end( + test.id(), "SKIP", message=reason, expected="PASS", extra=extra + ) + + +class StructuredTestRunner(unittest.TextTestRunner): + + resultclass = StructuredTestResult + + def __init__(self, **kwargs): + """TestRunner subclass designed for structured logging. + + :params logger: A ``StructuredLogger`` to use for logging the test run. + :params test_list: An optional list of tests that will be passed along + the `suite_start` message. + + """ + + self.logger = kwargs.pop("logger") + self.test_list = kwargs.pop("test_list", []) + self.result_callbacks = kwargs.pop("result_callbacks", []) + unittest.TextTestRunner.__init__(self, **kwargs) + + def _makeResult(self): + return self.resultclass( + self.stream, + self.descriptions, + self.verbosity, + logger=self.logger, + test_list=self.test_list, + ) + + def run(self, test): + """Run the given test case or test suite.""" + result = self._makeResult() + result.failfast = self.failfast + result.buffer = self.buffer + startTime = time.time() + startTestRun = getattr(result, "startTestRun", None) + if startTestRun is not None: + startTestRun() + try: + test(result) + finally: + stopTestRun = getattr(result, "stopTestRun", None) + if stopTestRun is not None: + stopTestRun() + stopTime = time.time() + if hasattr(result, "time_taken"): + result.time_taken = stopTime - startTime + + return result |