diff --git a/debian/copyright b/debian/copyright index 6d62e84c3bf..835ec78a6cb 100644 --- a/debian/copyright +++ b/debian/copyright @@ -305,6 +305,10 @@ Files: addons/web/static/lib/popper/* Copyright: 2016 Federico Zivolo License: MIT +Files: odoo/odoo/tests/case.py odoo/odoo/tests/result.py odoo/odoo/tests/suite.py +Copyright: (c) 1999-2003 Steve Purcell; (c) 2003-2010 Python Software Foundation +License: PSF + Files: addons/web/static/lib/py.js/* Copyright: 2012 License: DWTFYW diff --git a/odoo/addons/base/tests/test_mimetypes.py b/odoo/addons/base/tests/test_mimetypes.py index fc393627c1a..fc9257b5ef4 100644 --- a/odoo/addons/base/tests/test_mimetypes.py +++ b/odoo/addons/base/tests/test_mimetypes.py @@ -126,8 +126,3 @@ class test_guess_mimetype(BaseCase): self.assertEqual(get_extension('filename.not_alnum'), '') self.assertEqual(get_extension('filename.with space'), '') self.assertEqual(get_extension('filename.notAnExtension'), '') - - - -if __name__ == '__main__': - unittest.main() diff --git a/odoo/addons/base/tests/test_test_suite.py b/odoo/addons/base/tests/test_test_suite.py index 8454b5f6d3c..307208b5ced 100644 --- a/odoo/addons/base/tests/test_test_suite.py +++ b/odoo/addons/base/tests/test_test_suite.py @@ -4,23 +4,29 @@ import difflib import logging import re +import sys from contextlib import contextmanager -from unittest import TestCase +from unittest import SkipTest, skip from unittest.mock import patch -from odoo.tests.common import TransactionCase +from odoo.tests.common import BaseCase, TransactionCase from odoo.tests.common import users, warmup -from odoo.tests.runner import OdooTestResult +from odoo.tests.result import OdooTestResult +from odoo.tests.case import TestCase _logger = logging.getLogger(__name__) from odoo.tests import MetaCase -class TestTestSuite(TestCase, metaclass=MetaCase): +if sys.version_info >= (3, 8): + # this is mainly to ensure that simple tests will continue to work even if BaseCase should be used + # this only works if doClassCleanup is available on testCase because of the vendoring of suite.py. + # this test will only work in python 3.8 + + class TestTestSuite(TestCase, metaclass=MetaCase): - def test_test_suite(self): - """ Check that OdooSuite handles unittest.TestCase correctly. """ + def test_test_suite(self): + """ Check that OdooSuite handles unittest.TestCase correctly. """ class TestRunnerLoggingCommon(TransactionCase): @@ -37,7 +43,7 @@ class TestRunnerLoggingCommon(TransactionCase): self.expected_first_frame_methods = None return super().setUp() - def _feedErrorsToResult(self, result, errors): + def _addError(self, result, test, exc_info): # We use this hook to catch the logged error. It is initially called # post tearDown, and logs the actual errors. Because of our hack # tests.common._ErrorCatcher, the errors are logged directly. This is @@ -48,11 +54,10 @@ class TestRunnerLoggingCommon(TransactionCase): self.test_result = result # while we are here, let's check that the first frame of the stack # is always inside the test method - for error in errors: - _, exc_info = error - if exc_info: - tb = exc_info[2] - self._check_first_frame(tb) + + if exc_info: + tb = exc_info[2] + self._check_first_frame(tb) # intercept all ir_logging. We cannot use log catchers or other # fancy stuff because makeRecord is too low level. @@ -70,7 +75,7 @@ class TestRunnerLoggingCommon(TransactionCase): fake_result = OdooTestResult() with patch('logging.Logger.makeRecord', makeRecord), patch('logging.Logger.handle', handle): - super()._feedErrorsToResult(fake_result, errors) + super()._addError(fake_result, test, exc_info) self._check_log_records(log_records) @@ -112,9 +117,9 @@ class TestRunnerLoggingCommon(TransactionCase): value = self._clean_message(value) if value != expected: if key != 'msg': - self._log_error(f"Key `{key}` => `{value}` is not equal to `{expected}` \n {log_record['str']}") + self._log_error(f"Key `{key}` => `{value}` is not equal to `{expected}` \n {log_record['msg']}") else: - diff = '\n'.join(difflib.ndiff(value.splitlines(), expected.splitlines())) + diff = '\n'.join(difflib.ndiff(expected.splitlines(), value.splitlines())) self._log_error(f"Key `{key}` did not matched expected:\n{diff}") def _log_error(self, message): @@ -134,6 +139,9 @@ class TestRunnerLoggingCommon(TransactionCase): class TestRunnerLogging(TestRunnerLoggingCommon): + def test_has_add_error(self): + self.assertTrue(hasattr(self, '_addError')) + def test_raise(self): raise Exception('This is an error') @@ -171,6 +179,8 @@ Exception: {message} @users('__system__') @warmup def test_with_decorators(self): + # note, this test may be broken with a decorator in decorator=5.0.5 since the behaviour changed + # but decoratorx was not introduced yet. message = ( '''ERROR: Subtest TestRunnerLogging.test_with_decorators (login='__system__') Traceback (most recent call last): @@ -448,3 +458,72 @@ class TestRunnerLoggingTeardown(TestRunnerLoggingCommon): with self.subTest(): raise Exception('This is a second subTest error') raise Exception('This is a test error') + + +class TestSubtests(BaseCase): + + def test_nested_subtests(self): + with self.subTest(a=1, x=2): + with self.subTest(b=3, x=4): + self.assertEqual(self._subtest._subDescription(), '(b=3, x=4, a=1)') + with self.subTest(b=5, x=6): + self.assertEqual(self._subtest._subDescription(), '(b=5, x=6, a=1)') + + +class TestClassSetup(BaseCase): + + @classmethod + def setUpClass(cls): + raise SkipTest('Skip this class') + + def test_method(self): + pass + + +class TestClassTeardown(BaseCase): + + @classmethod + def tearDownClass(cls): + raise SkipTest('Skip this class') + + def test_method(self): + pass + + +class Test01ClassCleanups(BaseCase): + """ + The purpose of this test combined with Test02ClassCleanupsCheck is to check that + class cleanup work. class cleanup where introduced in python3.8 but tests should + remain compatible with python 3.7 + """ + executed = False + cleanup = False + + @classmethod + def setUpClass(cls): + cls.executed = True + + def doCleanup(): + cls.cleanup = True + cls.addClassCleanup(doCleanup) + + def test_dummy(self): + pass + + +class Test02ClassCleanupsCheck(BaseCase): + def test_classcleanups(self): + self.assertTrue(Test01ClassCleanups.executed, "This test only makes sence when executed after Test01ClassCleanups") + self.assertTrue(Test01ClassCleanups.cleanup, "TestClassCleanup shoudl have been cleanuped") + + +@skip +class TestSkipClass(BaseCase): + def test_classcleanups(self): + raise Exception('This should be skipped') + + +class TestSkipMethof(BaseCase): + @skip + def test_skip_method(self): + raise Exception('This should be skipped') diff --git a/odoo/addons/base/tests/test_tests_tags.py b/odoo/addons/base/tests/test_tests_tags.py index ab230823cc4..e30678872fe 100644 --- a/odoo/addons/base/tests/test_tests_tags.py +++ b/odoo/addons/base/tests/test_tests_tags.py @@ -1,7 +1,8 @@ # -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. -from odoo.tests.common import TransactionCase, tagged, TagsSelector, BaseCase +from odoo.tests.common import TransactionCase, tagged, BaseCase +from odoo.tests.tag_selector import TagsSelector @tagged('nodatabase') diff --git a/odoo/modules/loading.py b/odoo/modules/loading.py index e805a1781ab..bda9f35fab9 100644 --- a/odoo/modules/loading.py +++ b/odoo/modules/loading.py @@ -320,7 +320,7 @@ def load_module_graph(env, graph, status=None, perform_checks=True, if test_results and not test_results.wasSuccessful(): _logger.error( "Module %s: %d failures, %d errors of %d tests", - module_name, len(test_results.failures), len(test_results.errors), + module_name, test_results.failures_count, test_results.errors_count, test_results.testsRun ) diff --git a/odoo/modules/registry.py b/odoo/modules/registry.py index 61e5da4973c..9ef7fe5177f 100644 --- a/odoo/modules/registry.py +++ b/odoo/modules/registry.py @@ -114,7 +114,7 @@ class Registry(Mapping): self._sql_constraints = set() self._init = True self._database_translated_fields = () # names of translated fields in database - self._assertion_report = odoo.tests.runner.OdooTestResult() + self._assertion_report = odoo.tests.result.OdooTestResult() self._fields_by_model = None self._ordinary_tables = None self._constraint_queue = deque() diff --git a/odoo/service/server.py b/odoo/service/server.py index c0d59a5acc7..e7b4a210fad 100644 --- a/odoo/service/server.py +++ b/odoo/service/server.py @@ -22,6 +22,8 @@ import psutil import werkzeug.serving from werkzeug.debug import DebuggedApplication +from ..tests import loader + if os.name == 'posix': # Unix only for workers import fcntl @@ -58,7 +60,6 @@ from odoo.modules.registry import Registry from odoo.release import nt_service_name from odoo.tools import config from odoo.tools import stripped_sys_argv, dumpstacks, log_ormcache_stats -from ..tests import loader, runner _logger = logging.getLogger(__name__) @@ -579,7 +580,7 @@ class ThreadedServer(CommonServer): if stop: if config['test_enable']: - logger = odoo.tests.runner._logger + logger = odoo.tests.result._logger with Registry.registries._lock: for db, registry in Registry.registries.d.items(): report = registry._assertion_report @@ -1267,7 +1268,8 @@ def _reexec(updated_modules=None): os.execve(sys.executable, args, os.environ) def load_test_file_py(registry, test_file): - from odoo.tests.common import OdooSuite + # pylint: disable=import-outside-toplevel + from odoo.tests.suite import OdooSuite threading.current_thread().testing = True try: test_path, _ = os.path.splitext(os.path.abspath(test_file)) diff --git a/odoo/tests/__init__.py b/odoo/tests/__init__.py index 5f95dd18a27..dbe824a3f23 100644 --- a/odoo/tests/__init__.py +++ b/odoo/tests/__init__.py @@ -1,3 +1,10 @@ +""" +Odoo unit testing framework, based on Python unittest. + +Some files as case.py, resut.py, suite.py are higly modified versions of unitest +See https://github.com/python/cpython/tree/3.10/Lib/unittest for reference files. +""" + from . import common from .common import * from . import test_parse_inline_template diff --git a/odoo/tests/case.py b/odoo/tests/case.py index 3223c0bff6f..3beb549a142 100644 --- a/odoo/tests/case.py +++ b/odoo/tests/case.py @@ -1,426 +1,102 @@ """Test case implementation""" import sys -import functools -import difflib +import inspect import logging -import pprint -import re -import warnings -import collections import contextlib -import traceback -import types +from unittest import SkipTest, TestCase as _TestCase + +_logger = logging.getLogger(__name__) -from . import result -from .util import (strclass, safe_repr, _count_diff_all_purpose, - _count_diff_hashable, _common_shorten_repr) __unittest = True _subtest_msg_sentinel = object() -DIFF_OMITTED = ('\nDiff is %s characters long. ' - 'Set self.maxDiff to None to see it.') - -class SkipTest(Exception): - """ - Raise this exception in a test to skip it. - - Usually you can use TestCase.skipTest() or one of the skipping decorators - instead of raising this directly. - """ - -class _ShouldStop(Exception): - """ - The test should stop. - """ - -class _UnexpectedSuccess(Exception): - """ - The test was supposed to fail, but it didn't! - """ - class _Outcome(object): - def __init__(self, result=None): - self.expecting_failure = False + def __init__(self, test, result): self.result = result - self.result_supports_subtests = hasattr(result, "addSubTest") self.success = True - self.skipped = [] - self.expectedFailure = None - self.errors = [] + self.test = test @contextlib.contextmanager def testPartExecutor(self, test_case, isTest=False): - old_success = self.success - self.success = True try: yield except KeyboardInterrupt: raise except SkipTest as e: self.success = False - self.skipped.append((test_case, str(e))) - except _ShouldStop: - pass - except: + self.result.addSkip(test_case, str(e)) + except: # pylint: disable=bare-except exc_info = sys.exc_info() - if self.expecting_failure: - self.expectedFailure = exc_info - else: - self.success = False - self.errors.append((test_case, exc_info)) + self.success = False + + if exc_info is not None: + exception_type, exception, tb = exc_info + tb = self._complete_traceback(tb) + exc_info = (exception_type, exception, tb) + self.test._addError(self.result, test_case, exc_info) + # explicitly break a reference cycle: # exc_info -> frame -> exc_info exc_info = None + + def _complete_traceback(self, initial_tb): + Traceback = type(initial_tb) + + # make the set of frames in the traceback + tb_frames = set() + tb = initial_tb + while tb: + tb_frames.add(tb.tb_frame) + tb = tb.tb_next + tb = initial_tb + + # find the common frame by searching the last frame of the current_stack present in the traceback. + current_frame = inspect.currentframe() + common_frame = None + while current_frame: + if current_frame in tb_frames: + common_frame = current_frame # we want to find the last frame in common + current_frame = current_frame.f_back + + if not common_frame: # not really useful but safer + _logger.warning('No common frame found with current stack, displaying full stack') + tb = initial_tb else: - if self.result_supports_subtests and self.success: - self.errors.append((test_case, None)) - finally: - self.success = self.success and old_success + # remove the tb_frames until the common_frame is reached (keep the current_frame tb since the line is more accurate) + while tb and tb.tb_frame != common_frame: + tb = tb.tb_next + + # add all current frame elements under the common_frame to tb + current_frame = common_frame.f_back + while current_frame: + tb = Traceback(tb, current_frame, current_frame.f_lasti, current_frame.f_lineno) + current_frame = current_frame.f_back + + # remove traceback root part (odoo_bin, main, loading, ...), as + # everything under the testCase is not useful. Using '_callTestMethod', + # '_callSetUp', '_callTearDown', '_callCleanup' instead of the test + # method since the error does not comme especially from the test method. + while tb: + code = tb.tb_frame.f_code + if code.co_filename.endswith('/case.py') and code.co_name in ('_callTestMethod', '_callSetUp', '_callTearDown', '_callCleanup'): + return tb.tb_next + tb = tb.tb_next + + _logger.warning('No root frame found, displaying full stacks') + return initial_tb # this shouldn't be reached -def _id(obj): - return obj - - -_module_cleanups = [] -def addModuleCleanup(function, /, *args, **kwargs): - """Same as addCleanup, except the cleanup items are called even if - setUpModule fails (unlike tearDownModule).""" - _module_cleanups.append((function, args, kwargs)) - - -def doModuleCleanups(): - """Execute all module cleanup functions. Normally called for you after - tearDownModule.""" - exceptions = [] - while _module_cleanups: - function, args, kwargs = _module_cleanups.pop() - try: - function(*args, **kwargs) - except Exception as exc: - exceptions.append(exc) - if exceptions: - # Swallows all but first exception. If a multi-exception handler - # gets written we should use that here instead. - raise exceptions[0] - - -def skip(reason): - """ - Unconditionally skip a test. - """ - def decorator(test_item): - if not isinstance(test_item, type): - @functools.wraps(test_item) - def skip_wrapper(*args, **kwargs): - raise SkipTest(reason) - test_item = skip_wrapper - - test_item.__unittest_skip__ = True - test_item.__unittest_skip_why__ = reason - return test_item - if isinstance(reason, types.FunctionType): - test_item = reason - reason = '' - return decorator(test_item) - return decorator - -def skipIf(condition, reason): - """ - Skip a test if the condition is true. - """ - if condition: - return skip(reason) - return _id - -def skipUnless(condition, reason): - """ - Skip a test unless the condition is true. - """ - if not condition: - return skip(reason) - return _id - -def expectedFailure(test_item): - test_item.__unittest_expecting_failure__ = True - return test_item - -def _is_subtype(expected, basetype): - if isinstance(expected, tuple): - return all(_is_subtype(e, basetype) for e in expected) - return isinstance(expected, type) and issubclass(expected, basetype) - -class _BaseTestCaseContext: - - def __init__(self, test_case): - self.test_case = test_case - - def _raiseFailure(self, standardMsg): - msg = self.test_case._formatMessage(self.msg, standardMsg) - raise self.test_case.failureException(msg) - -class _AssertRaisesBaseContext(_BaseTestCaseContext): - - def __init__(self, expected, test_case, expected_regex=None): - _BaseTestCaseContext.__init__(self, test_case) - self.expected = expected - self.test_case = test_case - if expected_regex is not None: - expected_regex = re.compile(expected_regex) - self.expected_regex = expected_regex - self.obj_name = None - self.msg = None - - def handle(self, name, args, kwargs): - """ - If args is empty, assertRaises/Warns is being used as a - context manager, so check for a 'msg' kwarg and return self. - If args is not empty, call a callable passing positional and keyword - arguments. - """ - try: - if not _is_subtype(self.expected, self._base_type): - raise TypeError('%s() arg 1 must be %s' % - (name, self._base_type_str)) - if not args: - self.msg = kwargs.pop('msg', None) - if kwargs: - raise TypeError('%r is an invalid keyword argument for ' - 'this function' % (next(iter(kwargs)),)) - return self - - callable_obj, *args = args - try: - self.obj_name = callable_obj.__name__ - except AttributeError: - self.obj_name = str(callable_obj) - with self: - callable_obj(*args, **kwargs) - finally: - # bpo-23890: manually break a reference cycle - self = None - - -class _AssertRaisesContext(_AssertRaisesBaseContext): - """A context manager used to implement TestCase.assertRaises* methods.""" - - _base_type = BaseException - _base_type_str = 'an exception type or tuple of exception types' - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_value, tb): - if exc_type is None: - try: - exc_name = self.expected.__name__ - except AttributeError: - exc_name = str(self.expected) - if self.obj_name: - self._raiseFailure("{} not raised by {}".format(exc_name, - self.obj_name)) - else: - self._raiseFailure("{} not raised".format(exc_name)) - else: - traceback.clear_frames(tb) - if not issubclass(exc_type, self.expected): - # let unexpected exceptions pass through - return False - # store exception, without traceback, for later retrieval - self.exception = exc_value.with_traceback(None) - if self.expected_regex is None: - return True - - expected_regex = self.expected_regex - if not expected_regex.search(str(exc_value)): - self._raiseFailure('"{}" does not match "{}"'.format( - expected_regex.pattern, str(exc_value))) - return True - - -class _AssertWarnsContext(_AssertRaisesBaseContext): - """A context manager used to implement TestCase.assertWarns* methods.""" - - _base_type = Warning - _base_type_str = 'a warning type or tuple of warning types' - - def __enter__(self): - # The __warningregistry__'s need to be in a pristine state for tests - # to work properly. - for v in list(sys.modules.values()): - if getattr(v, '__warningregistry__', None): - v.__warningregistry__ = {} - self.warnings_manager = warnings.catch_warnings(record=True) - self.warnings = self.warnings_manager.__enter__() - warnings.simplefilter("always", self.expected) - return self - - def __exit__(self, exc_type, exc_value, tb): - self.warnings_manager.__exit__(exc_type, exc_value, tb) - if exc_type is not None: - # let unexpected exceptions pass through - return - try: - exc_name = self.expected.__name__ - except AttributeError: - exc_name = str(self.expected) - first_matching = None - for m in self.warnings: - w = m.message - if not isinstance(w, self.expected): - continue - if first_matching is None: - first_matching = w - if (self.expected_regex is not None and - not self.expected_regex.search(str(w))): - continue - # store warning for later retrieval - self.warning = w - self.filename = m.filename - self.lineno = m.lineno - return - # Now we simply try to choose a helpful failure message - if first_matching is not None: - self._raiseFailure('"{}" does not match "{}"'.format( - self.expected_regex.pattern, str(first_matching))) - if self.obj_name: - self._raiseFailure("{} not triggered by {}".format(exc_name, - self.obj_name)) - else: - self._raiseFailure("{} not triggered".format(exc_name)) - - - -_LoggingWatcher = collections.namedtuple("_LoggingWatcher", - ["records", "output"]) - - -class _CapturingHandler(logging.Handler): - """ - A logging handler capturing all (raw and formatted) logging output. - """ - - def __init__(self): - logging.Handler.__init__(self) - self.watcher = _LoggingWatcher([], []) - - def flush(self): - pass - - def emit(self, record): - self.watcher.records.append(record) - msg = self.format(record) - self.watcher.output.append(msg) - - - -class _AssertLogsContext(_BaseTestCaseContext): - """A context manager used to implement TestCase.assertLogs().""" - - LOGGING_FORMAT = "%(levelname)s:%(name)s:%(message)s" - - def __init__(self, test_case, logger_name, level): - _BaseTestCaseContext.__init__(self, test_case) - self.logger_name = logger_name - if level: - self.level = logging._nameToLevel.get(level, level) - else: - self.level = logging.INFO - self.msg = None - - def __enter__(self): - if isinstance(self.logger_name, logging.Logger): - logger = self.logger = self.logger_name - else: - logger = self.logger = logging.getLogger(self.logger_name) - formatter = logging.Formatter(self.LOGGING_FORMAT) - handler = _CapturingHandler() - handler.setFormatter(formatter) - self.watcher = handler.watcher - self.old_handlers = logger.handlers[:] - self.old_level = logger.level - self.old_propagate = logger.propagate - logger.handlers = [handler] - logger.setLevel(self.level) - logger.propagate = False - return handler.watcher - - def __exit__(self, exc_type, exc_value, tb): - self.logger.handlers = self.old_handlers - self.logger.propagate = self.old_propagate - self.logger.setLevel(self.old_level) - if exc_type is not None: - # let unexpected exceptions pass through - return False - if len(self.watcher.records) == 0: - self._raiseFailure( - "no logs of level {} or higher triggered on {}" - .format(logging.getLevelName(self.level), self.logger.name)) - - -class _OrderedChainMap(collections.ChainMap): - def __iter__(self): - seen = set() - for mapping in self.maps: - for k in mapping: - if k not in seen: - seen.add(k) - yield k - - -class TestCase(object): - """A class whose instances are single test cases. - - By default, the test code itself should be placed in a method named - 'runTest'. - - If the fixture may be used for many test cases, create as - many test methods as are needed. When instantiating such a TestCase - subclass, specify in the constructor arguments the name of the test method - that the instance is to execute. - - Test authors should subclass TestCase for their own tests. Construction - and deconstruction of the test's environment ('fixture') can be - implemented by overriding the 'setUp' and 'tearDown' methods respectively. - - If it is necessary to override the __init__ method, the base class - __init__ method must always be called. It is important that subclasses - should not change the signature of their __init__ method, since instances - of the classes are instantiated automatically by parts of the framework - in order to be run. - - When subclassing TestCase, you can set these attributes: - * failureException: determines which exception will be raised when - the instance's assertion methods fail; test methods raising this - exception will be deemed to have 'failed' rather than 'errored'. - * longMessage: determines whether long messages (including repr of - objects used in assert methods) will be printed on failure in *addition* - to any explicit message passed. - * maxDiff: sets the maximum length of a diff in failure messages - by assert methods using difflib. It is looked up as an instance - attribute so can be configured by individual tests if required. - """ - - failureException = AssertionError - - longMessage = True - - maxDiff = 80*8 - - # If a string is longer than _diffThreshold, use normal comparison instead - # of difflib. See #11763. - _diffThreshold = 2**16 - - # Attribute used by TestSuite for classSetUp - - _classSetupFailed = False - - _class_cleanups = [] +class TestCase(_TestCase): + _class_cleanups = [] # needed, backport for versions < 3.8 + __unittest_skip__ = False + __unittest_skip_why__ = '' + _moduleSetUpFailed = False + # pylint: disable=super-init-not-called def __init__(self, methodName='runTest'): """Create an instance of the class that will use the named test method when executed. Raises a ValueError if the instance does @@ -428,17 +104,11 @@ class TestCase(object): """ self._testMethodName = methodName self._outcome = None - self._testMethodDoc = 'No test' - try: - testMethod = getattr(self, methodName) - except AttributeError: - if methodName != 'runTest': - # we allow instantiation with no explicit method name - # but not an *incorrect* or missing method name - raise ValueError("no such test method in %s: %s" % - (self.__class__, methodName)) - else: - self._testMethodDoc = testMethod.__doc__ + if methodName != 'runTest' and not hasattr(self, methodName): + # we allow instantiation with no explicit method name + # but not an *incorrect* or missing method name + raise ValueError("no such test method in %s: %s" % + (self.__class__, methodName)) self._cleanups = [] self._subtest = None @@ -453,112 +123,22 @@ class TestCase(object): self.addTypeEqualityFunc(frozenset, 'assertSetEqual') self.addTypeEqualityFunc(str, 'assertMultiLineEqual') - def addTypeEqualityFunc(self, typeobj, function): - """Add a type specific assertEqual style function to compare a type. - - This method is for use by TestCase subclasses that need to register - their own type equality functions to provide nicer error messages. - - Args: - typeobj: The data type to call this function on when both values - are of the same type in assertEqual(). - function: The callable taking two arguments and an optional - msg= argument that raises self.failureException with a - useful error message when the two arguments are not equal. - """ - self._type_equality_funcs[typeobj] = function - - def addCleanup(*args, **kwargs): + def addCleanup(self, function, *args, **kwargs): """Add a function, with arguments, to be called when the test is completed. Functions added are called on a LIFO basis and are called after tearDown on test failure or success. Cleanup items are called even if setUp fails (unlike tearDown).""" - if len(args) >= 2: - self, function, *args = args - elif not args: - raise TypeError("descriptor 'addCleanup' of 'TestCase' object " - "needs an argument") - elif 'function' in kwargs: - function = kwargs.pop('function') - self, *args = args - import warnings - warnings.warn("Passing 'function' as keyword argument is deprecated", - DeprecationWarning, stacklevel=2) - else: - raise TypeError('addCleanup expected at least 1 positional ' - 'argument, got %d' % (len(args)-1)) - args = tuple(args) - self._cleanups.append((function, args, kwargs)) - addCleanup.__text_signature__ = '($self, function, /, *args, **kwargs)' @classmethod - def addClassCleanup(cls, function, /, *args, **kwargs): + def addClassCleanup(cls, function, *args, **kwargs): """Same as addCleanup, except the cleanup items are called even if setUpClass fails (unlike tearDownClass).""" cls._class_cleanups.append((function, args, kwargs)) - def setUp(self): - "Hook method for setting up the test fixture before exercising it." - pass - - def tearDown(self): - "Hook method for deconstructing the test fixture after testing it." - pass - - @classmethod - def setUpClass(cls): - "Hook method for setting up class fixture before running tests in the class." - - @classmethod - def tearDownClass(cls): - "Hook method for deconstructing the class fixture after running all tests in the class." - - def countTestCases(self): - return 1 - - def defaultTestResult(self): - return result.TestResult() - def shortDescription(self): - """Returns a one-line description of the test, or None if no - description has been provided. - - The default implementation of this method returns the first line of - the specified test method's docstring. - """ - doc = self._testMethodDoc - return doc.strip().split("\n")[0].strip() if doc else None - - - def id(self): - return "%s.%s" % (strclass(self.__class__), self._testMethodName) - - def __eq__(self, other): - if type(self) is not type(other): - return NotImplemented - - return self._testMethodName == other._testMethodName - - def __hash__(self): - return hash((type(self), self._testMethodName)) - - def __str__(self): - return "%s (%s)" % (self._testMethodName, strclass(self.__class__)) - - def __repr__(self): - return "<%s testMethod=%s>" % \ - (strclass(self.__class__), self._testMethodName) - - def _addSkip(self, result, test_case, reason): - addSkip = getattr(result, 'addSkip', None) - if addSkip is not None: - addSkip(test_case, reason) - else: - warnings.warn("TestResult has no addSkip method, skips not reported", - RuntimeWarning, 2) - result.addSuccess(test_case) + return None @contextlib.contextmanager def subTest(self, msg=_subtest_msg_sentinel, **params): @@ -568,63 +148,32 @@ class TestCase(object): case as failed but resumes execution at the end of the enclosed block, allowing further test code to be executed. """ - if self._outcome is None or not self._outcome.result_supports_subtests: - yield - return parent = self._subtest - if parent is None: - params_map = _OrderedChainMap(params) - else: - params_map = parent.params.new_child(params) - self._subtest = _SubTest(self, msg, params_map) + if parent: + params = {**params, **{k: v for k, v in parent.params.items() if k not in params}} + self._subtest = _SubTest(self, msg, params) try: with self._outcome.testPartExecutor(self._subtest, isTest=True): yield - if not self._outcome.success: - result = self._outcome.result - if result is not None and result.failfast: - raise _ShouldStop - elif self._outcome.expectedFailure: - # If the test is expecting a failure, we really want to - # stop now and register the expected failure. - raise _ShouldStop finally: self._subtest = parent - def _feedErrorsToResult(self, result, errors): - for test, exc_info in errors: - if isinstance(test, _SubTest): - result.addSubTest(test.test_case, test, exc_info) - elif exc_info is not None: - if issubclass(exc_info[0], self.failureException): - result.addFailure(test, exc_info) - else: - result.addError(test, exc_info) + def _addError(self, result, test, exc_info): + """ + This method is similar to feed_errors_to_result in python<=3.10 + but only manage one error at a time + This is also inspired from python 3.11 _addError but still manages + subtests errors as in python 3.7-3.10 for minimal changes. + The method remains on the test to easily override it in test_test_suite - def _addExpectedFailure(self, result, exc_info): - try: - addExpectedFailure = result.addExpectedFailure - except AttributeError: - warnings.warn("TestResult has no addExpectedFailure method, reporting as passes", - RuntimeWarning) - result.addSuccess(self) - else: - addExpectedFailure(self, exc_info) - - def _addUnexpectedSuccess(self, result): - try: - addUnexpectedSuccess = result.addUnexpectedSuccess - except AttributeError: - warnings.warn("TestResult has no addUnexpectedSuccess method, reporting as failure", - RuntimeWarning) - # We need to pass an actual exception and traceback to addFailure, - # otherwise the legacy result can choke. - try: - raise _UnexpectedSuccess from None - except _UnexpectedSuccess: - result.addFailure(self, sys.exc_info()) - else: - addUnexpectedSuccess(self) + """ + if isinstance(test, _SubTest): + result.addSubTest(test.test_case, test, exc_info) + elif exc_info is not None: + if issubclass(exc_info[0], self.failureException): + result.addFailure(test, exc_info) + else: + result.addError(test, exc_info) def _callSetUp(self): self.setUp() @@ -635,74 +184,43 @@ class TestCase(object): def _callTearDown(self): self.tearDown() - def _callCleanup(self, function, /, *args, **kwargs): + def _callCleanup(self, function, *args, **kwargs): function(*args, **kwargs) - def run(self, result=None): - orig_result = result - if result is None: - result = self.defaultTestResult() - startTestRun = getattr(result, 'startTestRun', None) - if startTestRun is not None: - startTestRun() - + def run(self, result): result.startTest(self) testMethod = getattr(self, self._testMethodName) - if (getattr(self.__class__, "__unittest_skip__", False) or - getattr(testMethod, "__unittest_skip__", False)): - # If the class or method was skipped. - try: - skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') - or getattr(testMethod, '__unittest_skip_why__', '')) - self._addSkip(result, self, skip_why) - finally: - result.stopTest(self) + + skip = False + skip_why = '' + try: + skip = self.__class__.__unittest_skip__ or testMethod.__unittest_skip__ + skip_why = self.__class__.__unittest_skip_why__ or testMethod.__unittest_skip_why__ or '' + except AttributeError: # testMethod may not have a __unittest_skip__ or __unittest_skip_why__ + pass + if skip: + result.addSkip(self, skip_why) + result.stopTest(self) return - expecting_failure_method = getattr(testMethod, - "__unittest_expecting_failure__", False) - expecting_failure_class = getattr(self, - "__unittest_expecting_failure__", False) - expecting_failure = expecting_failure_class or expecting_failure_method - outcome = _Outcome(result) + + outcome = _Outcome(self, result) try: self._outcome = outcome - with outcome.testPartExecutor(self): self._callSetUp() if outcome.success: - outcome.expecting_failure = expecting_failure with outcome.testPartExecutor(self, isTest=True): self._callTestMethod(testMethod) - outcome.expecting_failure = False with outcome.testPartExecutor(self): self._callTearDown() self.doCleanups() - for test, reason in outcome.skipped: - self._addSkip(result, test, reason) - self._feedErrorsToResult(result, outcome.errors) if outcome.success: - if expecting_failure: - if outcome.expectedFailure: - self._addExpectedFailure(result, outcome.expectedFailure) - else: - self._addUnexpectedSuccess(result) - else: - result.addSuccess(self) + result.addSuccess(self) return result finally: result.stopTest(self) - if orig_result is None: - stopTestRun = getattr(result, 'stopTestRun', None) - if stopTestRun is not None: - stopTestRun() - - # explicitly break reference cycles: - # outcome.errors -> frame -> outcome -> outcome.errors - # outcome.expectedFailure -> frame -> outcome -> outcome.expectedFailure - outcome.errors.clear() - outcome.expectedFailure = None # clear the outcome, no more needed self._outcome = None @@ -710,16 +228,12 @@ class TestCase(object): def doCleanups(self): """Execute all cleanup functions. Normally called for you after tearDown.""" - outcome = self._outcome or _Outcome() + while self._cleanups: function, args, kwargs = self._cleanups.pop() - with outcome.testPartExecutor(self): + with self._outcome.testPartExecutor(self): self._callCleanup(function, *args, **kwargs) - # return this for backwards compatibility - # even though we no longer use it internally - return outcome.success - @classmethod def doClassCleanups(cls): """Execute all class cleanup functions. Normally called for you after @@ -729,758 +243,9 @@ class TestCase(object): function, args, kwargs = cls._class_cleanups.pop() try: function(*args, **kwargs) - except Exception as exc: + except Exception: cls.tearDown_exceptions.append(sys.exc_info()) - def __call__(self, *args, **kwds): - return self.run(*args, **kwds) - - def debug(self): - """Run the test without collecting errors in a TestResult""" - self.setUp() - getattr(self, self._testMethodName)() - self.tearDown() - while self._cleanups: - function, args, kwargs = self._cleanups.pop(-1) - function(*args, **kwargs) - - def skipTest(self, reason): - """Skip this test.""" - raise SkipTest(reason) - - def fail(self, msg=None): - """Fail immediately, with the given message.""" - raise self.failureException(msg) - - def assertFalse(self, expr, msg=None): - """Check that the expression is false.""" - if expr: - msg = self._formatMessage(msg, "%s is not false" % safe_repr(expr)) - raise self.failureException(msg) - - def assertTrue(self, expr, msg=None): - """Check that the expression is true.""" - if not expr: - msg = self._formatMessage(msg, "%s is not true" % safe_repr(expr)) - raise self.failureException(msg) - - def _formatMessage(self, msg, standardMsg): - """Honour the longMessage attribute when generating failure messages. - If longMessage is False this means: - * Use only an explicit message if it is provided - * Otherwise use the standard message for the assert - - If longMessage is True: - * Use the standard message - * If an explicit message is provided, plus ' : ' and the explicit message - """ - if not self.longMessage: - return msg or standardMsg - if msg is None: - return standardMsg - try: - # don't switch to '{}' formatting in Python 2.X - # it changes the way unicode input is handled - return '%s : %s' % (standardMsg, msg) - except UnicodeDecodeError: - return '%s : %s' % (safe_repr(standardMsg), safe_repr(msg)) - - def assertRaises(self, expected_exception, *args, **kwargs): - """Fail unless an exception of class expected_exception is raised - by the callable when invoked with specified positional and - keyword arguments. If a different type of exception is - raised, it will not be caught, and the test case will be - deemed to have suffered an error, exactly as for an - unexpected exception. - - If called with the callable and arguments omitted, will return a - context object used like this:: - - with self.assertRaises(SomeException): - do_something() - - An optional keyword argument 'msg' can be provided when assertRaises - is used as a context object. - - The context manager keeps a reference to the exception as - the 'exception' attribute. This allows you to inspect the - exception after the assertion:: - - with self.assertRaises(SomeException) as cm: - do_something() - the_exception = cm.exception - self.assertEqual(the_exception.error_code, 3) - """ - context = _AssertRaisesContext(expected_exception, self) - try: - return context.handle('assertRaises', args, kwargs) - finally: - # bpo-23890: manually break a reference cycle - context = None - - def assertWarns(self, expected_warning, *args, **kwargs): - """Fail unless a warning of class warnClass is triggered - by the callable when invoked with specified positional and - keyword arguments. If a different type of warning is - triggered, it will not be handled: depending on the other - warning filtering rules in effect, it might be silenced, printed - out, or raised as an exception. - - If called with the callable and arguments omitted, will return a - context object used like this:: - - with self.assertWarns(SomeWarning): - do_something() - - An optional keyword argument 'msg' can be provided when assertWarns - is used as a context object. - - The context manager keeps a reference to the first matching - warning as the 'warning' attribute; similarly, the 'filename' - and 'lineno' attributes give you information about the line - of Python code from which the warning was triggered. - This allows you to inspect the warning after the assertion:: - - with self.assertWarns(SomeWarning) as cm: - do_something() - the_warning = cm.warning - self.assertEqual(the_warning.some_attribute, 147) - """ - context = _AssertWarnsContext(expected_warning, self) - return context.handle('assertWarns', args, kwargs) - - def assertLogs(self, logger=None, level=None): - """Fail unless a log message of level *level* or higher is emitted - on *logger_name* or its children. If omitted, *level* defaults to - INFO and *logger* defaults to the root logger. - - This method must be used as a context manager, and will yield - a recording object with two attributes: `output` and `records`. - At the end of the context manager, the `output` attribute will - be a list of the matching formatted log messages and the - `records` attribute will be a list of the corresponding LogRecord - objects. - - Example:: - - with self.assertLogs('foo', level='INFO') as cm: - logging.getLogger('foo').info('first message') - logging.getLogger('foo.bar').error('second message') - self.assertEqual(cm.output, ['INFO:foo:first message', - 'ERROR:foo.bar:second message']) - """ - return _AssertLogsContext(self, logger, level) - - def _getAssertEqualityFunc(self, first, second): - """Get a detailed comparison function for the types of the two args. - - Returns: A callable accepting (first, second, msg=None) that will - raise a failure exception if first != second with a useful human - readable error message for those types. - """ - # - # NOTE(gregory.p.smith): I considered isinstance(first, type(second)) - # and vice versa. I opted for the conservative approach in case - # subclasses are not intended to be compared in detail to their super - # class instances using a type equality func. This means testing - # subtypes won't automagically use the detailed comparison. Callers - # should use their type specific assertSpamEqual method to compare - # subclasses if the detailed comparison is desired and appropriate. - # See the discussion in http://bugs.python.org/issue2578. - # - if type(first) is type(second): - asserter = self._type_equality_funcs.get(type(first)) - if asserter is not None: - if isinstance(asserter, str): - asserter = getattr(self, asserter) - return asserter - - return self._baseAssertEqual - - def _baseAssertEqual(self, first, second, msg=None): - """The default assertEqual implementation, not type specific.""" - if not first == second: - standardMsg = '%s != %s' % _common_shorten_repr(first, second) - msg = self._formatMessage(msg, standardMsg) - raise self.failureException(msg) - - def assertEqual(self, first, second, msg=None): - """Fail if the two objects are unequal as determined by the '==' - operator. - """ - assertion_func = self._getAssertEqualityFunc(first, second) - assertion_func(first, second, msg=msg) - - def assertNotEqual(self, first, second, msg=None): - """Fail if the two objects are equal as determined by the '!=' - operator. - """ - if not first != second: - msg = self._formatMessage(msg, '%s == %s' % (safe_repr(first), - safe_repr(second))) - raise self.failureException(msg) - - def assertAlmostEqual(self, first, second, places=None, msg=None, - delta=None): - """Fail if the two objects are unequal as determined by their - difference rounded to the given number of decimal places - (default 7) and comparing to zero, or by comparing that the - difference between the two objects is more than the given - delta. - - Note that decimal places (from zero) are usually not the same - as significant digits (measured from the most significant digit). - - If the two objects compare equal then they will automatically - compare almost equal. - """ - if first == second: - # shortcut - return - if delta is not None and places is not None: - raise TypeError("specify delta or places not both") - - diff = abs(first - second) - if delta is not None: - if diff <= delta: - return - - standardMsg = '%s != %s within %s delta (%s difference)' % ( - safe_repr(first), - safe_repr(second), - safe_repr(delta), - safe_repr(diff)) - else: - if places is None: - places = 7 - - if round(diff, places) == 0: - return - - standardMsg = '%s != %s within %r places (%s difference)' % ( - safe_repr(first), - safe_repr(second), - places, - safe_repr(diff)) - msg = self._formatMessage(msg, standardMsg) - raise self.failureException(msg) - - def assertNotAlmostEqual(self, first, second, places=None, msg=None, - delta=None): - """Fail if the two objects are equal as determined by their - difference rounded to the given number of decimal places - (default 7) and comparing to zero, or by comparing that the - difference between the two objects is less than the given delta. - - Note that decimal places (from zero) are usually not the same - as significant digits (measured from the most significant digit). - - Objects that are equal automatically fail. - """ - if delta is not None and places is not None: - raise TypeError("specify delta or places not both") - diff = abs(first - second) - if delta is not None: - if not (first == second) and diff > delta: - return - standardMsg = '%s == %s within %s delta (%s difference)' % ( - safe_repr(first), - safe_repr(second), - safe_repr(delta), - safe_repr(diff)) - else: - if places is None: - places = 7 - if not (first == second) and round(diff, places) != 0: - return - standardMsg = '%s == %s within %r places' % (safe_repr(first), - safe_repr(second), - places) - - msg = self._formatMessage(msg, standardMsg) - raise self.failureException(msg) - - def assertSequenceEqual(self, seq1, seq2, msg=None, seq_type=None): - """An equality assertion for ordered sequences (like lists and tuples). - - For the purposes of this function, a valid ordered sequence type is one - which can be indexed, has a length, and has an equality operator. - - Args: - seq1: The first sequence to compare. - seq2: The second sequence to compare. - seq_type: The expected datatype of the sequences, or None if no - datatype should be enforced. - msg: Optional message to use on failure instead of a list of - differences. - """ - if seq_type is not None: - seq_type_name = seq_type.__name__ - if not isinstance(seq1, seq_type): - raise self.failureException('First sequence is not a %s: %s' - % (seq_type_name, safe_repr(seq1))) - if not isinstance(seq2, seq_type): - raise self.failureException('Second sequence is not a %s: %s' - % (seq_type_name, safe_repr(seq2))) - else: - seq_type_name = "sequence" - - differing = None - try: - len1 = len(seq1) - except (TypeError, NotImplementedError): - differing = 'First %s has no length. Non-sequence?' % ( - seq_type_name) - - if differing is None: - try: - len2 = len(seq2) - except (TypeError, NotImplementedError): - differing = 'Second %s has no length. Non-sequence?' % ( - seq_type_name) - - if differing is None: - if seq1 == seq2: - return - - differing = '%ss differ: %s != %s\n' % ( - (seq_type_name.capitalize(),) + - _common_shorten_repr(seq1, seq2)) - - for i in range(min(len1, len2)): - try: - item1 = seq1[i] - except (TypeError, IndexError, NotImplementedError): - differing += ('\nUnable to index element %d of first %s\n' % - (i, seq_type_name)) - break - - try: - item2 = seq2[i] - except (TypeError, IndexError, NotImplementedError): - differing += ('\nUnable to index element %d of second %s\n' % - (i, seq_type_name)) - break - - if item1 != item2: - differing += ('\nFirst differing element %d:\n%s\n%s\n' % - ((i,) + _common_shorten_repr(item1, item2))) - break - else: - if (len1 == len2 and seq_type is None and - type(seq1) != type(seq2)): - # The sequences are the same, but have differing types. - return - - if len1 > len2: - differing += ('\nFirst %s contains %d additional ' - 'elements.\n' % (seq_type_name, len1 - len2)) - try: - differing += ('First extra element %d:\n%s\n' % - (len2, safe_repr(seq1[len2]))) - except (TypeError, IndexError, NotImplementedError): - differing += ('Unable to index element %d ' - 'of first %s\n' % (len2, seq_type_name)) - elif len1 < len2: - differing += ('\nSecond %s contains %d additional ' - 'elements.\n' % (seq_type_name, len2 - len1)) - try: - differing += ('First extra element %d:\n%s\n' % - (len1, safe_repr(seq2[len1]))) - except (TypeError, IndexError, NotImplementedError): - differing += ('Unable to index element %d ' - 'of second %s\n' % (len1, seq_type_name)) - standardMsg = differing - diffMsg = '\n' + '\n'.join( - difflib.ndiff(pprint.pformat(seq1).splitlines(), - pprint.pformat(seq2).splitlines())) - - standardMsg = self._truncateMessage(standardMsg, diffMsg) - msg = self._formatMessage(msg, standardMsg) - self.fail(msg) - - def _truncateMessage(self, message, diff): - max_diff = self.maxDiff - if max_diff is None or len(diff) <= max_diff: - return message + diff - return message + (DIFF_OMITTED % len(diff)) - - def assertListEqual(self, list1, list2, msg=None): - """A list-specific equality assertion. - - Args: - list1: The first list to compare. - list2: The second list to compare. - msg: Optional message to use on failure instead of a list of - differences. - - """ - self.assertSequenceEqual(list1, list2, msg, seq_type=list) - - def assertTupleEqual(self, tuple1, tuple2, msg=None): - """A tuple-specific equality assertion. - - Args: - tuple1: The first tuple to compare. - tuple2: The second tuple to compare. - msg: Optional message to use on failure instead of a list of - differences. - """ - self.assertSequenceEqual(tuple1, tuple2, msg, seq_type=tuple) - - def assertSetEqual(self, set1, set2, msg=None): - """A set-specific equality assertion. - - Args: - set1: The first set to compare. - set2: The second set to compare. - msg: Optional message to use on failure instead of a list of - differences. - - assertSetEqual uses ducktyping to support different types of sets, and - is optimized for sets specifically (parameters must support a - difference method). - """ - try: - difference1 = set1.difference(set2) - except TypeError as e: - self.fail('invalid type when attempting set difference: %s' % e) - except AttributeError as e: - self.fail('first argument does not support set difference: %s' % e) - - try: - difference2 = set2.difference(set1) - except TypeError as e: - self.fail('invalid type when attempting set difference: %s' % e) - except AttributeError as e: - self.fail('second argument does not support set difference: %s' % e) - - if not (difference1 or difference2): - return - - lines = [] - if difference1: - lines.append('Items in the first set but not the second:') - for item in difference1: - lines.append(repr(item)) - if difference2: - lines.append('Items in the second set but not the first:') - for item in difference2: - lines.append(repr(item)) - - standardMsg = '\n'.join(lines) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIn(self, member, container, msg=None): - """Just like self.assertTrue(a in b), but with a nicer default message.""" - if member not in container: - standardMsg = '%s not found in %s' % (safe_repr(member), - safe_repr(container)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertNotIn(self, member, container, msg=None): - """Just like self.assertTrue(a not in b), but with a nicer default message.""" - if member in container: - standardMsg = '%s unexpectedly found in %s' % (safe_repr(member), - safe_repr(container)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIs(self, expr1, expr2, msg=None): - """Just like self.assertTrue(a is b), but with a nicer default message.""" - if expr1 is not expr2: - standardMsg = '%s is not %s' % (safe_repr(expr1), - safe_repr(expr2)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIsNot(self, expr1, expr2, msg=None): - """Just like self.assertTrue(a is not b), but with a nicer default message.""" - if expr1 is expr2: - standardMsg = 'unexpectedly identical: %s' % (safe_repr(expr1),) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertDictEqual(self, d1, d2, msg=None): - self.assertIsInstance(d1, dict, 'First argument is not a dictionary') - self.assertIsInstance(d2, dict, 'Second argument is not a dictionary') - - if d1 != d2: - standardMsg = '%s != %s' % _common_shorten_repr(d1, d2) - diff = ('\n' + '\n'.join(difflib.ndiff( - pprint.pformat(d1).splitlines(), - pprint.pformat(d2).splitlines()))) - standardMsg = self._truncateMessage(standardMsg, diff) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertDictContainsSubset(self, subset, dictionary, msg=None): - """Checks whether dictionary is a superset of subset.""" - warnings.warn('assertDictContainsSubset is deprecated', - DeprecationWarning) - missing = [] - mismatched = [] - for key, value in subset.items(): - if key not in dictionary: - missing.append(key) - elif value != dictionary[key]: - mismatched.append('%s, expected: %s, actual: %s' % - (safe_repr(key), safe_repr(value), - safe_repr(dictionary[key]))) - - if not (missing or mismatched): - return - - standardMsg = '' - if missing: - standardMsg = 'Missing: %s' % ','.join(safe_repr(m) for m in - missing) - if mismatched: - if standardMsg: - standardMsg += '; ' - standardMsg += 'Mismatched values: %s' % ','.join(mismatched) - - self.fail(self._formatMessage(msg, standardMsg)) - - - def assertCountEqual(self, first, second, msg=None): - """Asserts that two iterables have the same elements, the same number of - times, without regard to order. - - self.assertEqual(Counter(list(first)), - Counter(list(second))) - - Example: - - [0, 1, 1] and [1, 0, 1] compare equal. - - [0, 0, 1] and [0, 1] compare unequal. - - """ - first_seq, second_seq = list(first), list(second) - try: - first = collections.Counter(first_seq) - second = collections.Counter(second_seq) - except TypeError: - # Handle case with unhashable elements - differences = _count_diff_all_purpose(first_seq, second_seq) - else: - if first == second: - return - differences = _count_diff_hashable(first_seq, second_seq) - - if differences: - standardMsg = 'Element counts were not equal:\n' - lines = ['First has %d, Second has %d: %r' % diff for diff in differences] - diffMsg = '\n'.join(lines) - standardMsg = self._truncateMessage(standardMsg, diffMsg) - msg = self._formatMessage(msg, standardMsg) - self.fail(msg) - - def assertMultiLineEqual(self, first, second, msg=None): - """Assert that two multi-line strings are equal.""" - self.assertIsInstance(first, str, 'First argument is not a string') - self.assertIsInstance(second, str, 'Second argument is not a string') - - if first != second: - # don't use difflib if the strings are too long - if (len(first) > self._diffThreshold or - len(second) > self._diffThreshold): - self._baseAssertEqual(first, second, msg) - firstlines = first.splitlines(keepends=True) - secondlines = second.splitlines(keepends=True) - if len(firstlines) == 1 and first.strip('\r\n') == first: - firstlines = [first + '\n'] - secondlines = [second + '\n'] - standardMsg = '%s != %s' % _common_shorten_repr(first, second) - diff = '\n' + ''.join(difflib.ndiff(firstlines, secondlines)) - standardMsg = self._truncateMessage(standardMsg, diff) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertLess(self, a, b, msg=None): - """Just like self.assertTrue(a < b), but with a nicer default message.""" - if not a < b: - standardMsg = '%s not less than %s' % (safe_repr(a), safe_repr(b)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertLessEqual(self, a, b, msg=None): - """Just like self.assertTrue(a <= b), but with a nicer default message.""" - if not a <= b: - standardMsg = '%s not less than or equal to %s' % (safe_repr(a), safe_repr(b)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertGreater(self, a, b, msg=None): - """Just like self.assertTrue(a > b), but with a nicer default message.""" - if not a > b: - standardMsg = '%s not greater than %s' % (safe_repr(a), safe_repr(b)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertGreaterEqual(self, a, b, msg=None): - """Just like self.assertTrue(a >= b), but with a nicer default message.""" - if not a >= b: - standardMsg = '%s not greater than or equal to %s' % (safe_repr(a), safe_repr(b)) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIsNone(self, obj, msg=None): - """Same as self.assertTrue(obj is None), with a nicer default message.""" - if obj is not None: - standardMsg = '%s is not None' % (safe_repr(obj),) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIsNotNone(self, obj, msg=None): - """Included for symmetry with assertIsNone.""" - if obj is None: - standardMsg = 'unexpectedly None' - self.fail(self._formatMessage(msg, standardMsg)) - - def assertIsInstance(self, obj, cls, msg=None): - """Same as self.assertTrue(isinstance(obj, cls)), with a nicer - default message.""" - if not isinstance(obj, cls): - standardMsg = '%s is not an instance of %r' % (safe_repr(obj), cls) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertNotIsInstance(self, obj, cls, msg=None): - """Included for symmetry with assertIsInstance.""" - if isinstance(obj, cls): - standardMsg = '%s is an instance of %r' % (safe_repr(obj), cls) - self.fail(self._formatMessage(msg, standardMsg)) - - def assertRaisesRegex(self, expected_exception, expected_regex, - *args, **kwargs): - """Asserts that the message in a raised exception matches a regex. - - Args: - expected_exception: Exception class expected to be raised. - expected_regex: Regex (re.Pattern object or string) expected - to be found in error message. - args: Function to be called and extra positional args. - kwargs: Extra kwargs. - msg: Optional message used in case of failure. Can only be used - when assertRaisesRegex is used as a context manager. - """ - context = _AssertRaisesContext(expected_exception, self, expected_regex) - return context.handle('assertRaisesRegex', args, kwargs) - - def assertWarnsRegex(self, expected_warning, expected_regex, - *args, **kwargs): - """Asserts that the message in a triggered warning matches a regexp. - Basic functioning is similar to assertWarns() with the addition - that only warnings whose messages also match the regular expression - are considered successful matches. - - Args: - expected_warning: Warning class expected to be triggered. - expected_regex: Regex (re.Pattern object or string) expected - to be found in error message. - args: Function to be called and extra positional args. - kwargs: Extra kwargs. - msg: Optional message used in case of failure. Can only be used - when assertWarnsRegex is used as a context manager. - """ - context = _AssertWarnsContext(expected_warning, self, expected_regex) - return context.handle('assertWarnsRegex', args, kwargs) - - def assertRegex(self, text, expected_regex, msg=None): - """Fail the test unless the text matches the regular expression.""" - if isinstance(expected_regex, (str, bytes)): - assert expected_regex, "expected_regex must not be empty." - expected_regex = re.compile(expected_regex) - if not expected_regex.search(text): - standardMsg = "Regex didn't match: %r not found in %r" % ( - expected_regex.pattern, text) - # _formatMessage ensures the longMessage option is respected - msg = self._formatMessage(msg, standardMsg) - raise self.failureException(msg) - - def assertNotRegex(self, text, unexpected_regex, msg=None): - """Fail the test if the text matches the regular expression.""" - if isinstance(unexpected_regex, (str, bytes)): - unexpected_regex = re.compile(unexpected_regex) - match = unexpected_regex.search(text) - if match: - standardMsg = 'Regex matched: %r matches %r in %r' % ( - text[match.start() : match.end()], - unexpected_regex.pattern, - text) - # _formatMessage ensures the longMessage option is respected - msg = self._formatMessage(msg, standardMsg) - raise self.failureException(msg) - - - def _deprecate(original_func): - def deprecated_func(*args, **kwargs): - warnings.warn( - 'Please use {0} instead.'.format(original_func.__name__), - DeprecationWarning, 2) - return original_func(*args, **kwargs) - return deprecated_func - - # see #9424 - failUnlessEqual = assertEquals = _deprecate(assertEqual) - failIfEqual = assertNotEquals = _deprecate(assertNotEqual) - failUnlessAlmostEqual = assertAlmostEquals = _deprecate(assertAlmostEqual) - failIfAlmostEqual = assertNotAlmostEquals = _deprecate(assertNotAlmostEqual) - failUnless = assert_ = _deprecate(assertTrue) - failUnlessRaises = _deprecate(assertRaises) - failIf = _deprecate(assertFalse) - assertRaisesRegexp = _deprecate(assertRaisesRegex) - assertRegexpMatches = _deprecate(assertRegex) - assertNotRegexpMatches = _deprecate(assertNotRegex) - - - -class FunctionTestCase(TestCase): - """A test case that wraps a test function. - - This is useful for slipping pre-existing test functions into the - unittest framework. Optionally, set-up and tidy-up functions can be - supplied. As with TestCase, the tidy-up ('tearDown') function will - always be called if the set-up ('setUp') function ran successfully. - """ - - def __init__(self, testFunc, setUp=None, tearDown=None, description=None): - super(FunctionTestCase, self).__init__() - self._setUpFunc = setUp - self._tearDownFunc = tearDown - self._testFunc = testFunc - self._description = description - - def setUp(self): - if self._setUpFunc is not None: - self._setUpFunc() - - def tearDown(self): - if self._tearDownFunc is not None: - self._tearDownFunc() - - def runTest(self): - self._testFunc() - - def id(self): - return self._testFunc.__name__ - - def __eq__(self, other): - if not isinstance(other, self.__class__): - return NotImplemented - - return self._setUpFunc == other._setUpFunc and \ - self._tearDownFunc == other._tearDownFunc and \ - self._testFunc == other._testFunc and \ - self._description == other._description - - def __hash__(self): - return hash((type(self), self._setUpFunc, self._tearDownFunc, - self._testFunc, self._description)) - - def __str__(self): - return "%s (%s)" % (strclass(self.__class__), - self._testFunc.__name__) - - def __repr__(self): - return "<%s tec=%s>" % (strclass(self.__class__), - self._testFunc) - - def shortDescription(self): - if self._description is not None: - return self._description - doc = self._testFunc.__doc__ - return doc and doc.split("\n")[0].strip() or None - class _SubTest(TestCase): @@ -1508,11 +273,5 @@ class _SubTest(TestCase): def id(self): return "{} {}".format(self.test_case.id(), self._subDescription()) - def shortDescription(self): - """Returns a one-line description of the subtest, or None if no - description has been provided. - """ - return self.test_case.shortDescription() - def __str__(self): return "{} {}".format(self.test_case, self._subDescription()) diff --git a/odoo/tests/common.py b/odoo/tests/common.py index b07ef9bcd4d..94a910d4a9b 100644 --- a/odoo/tests/common.py +++ b/odoo/tests/common.py @@ -29,6 +29,7 @@ import tempfile import threading import time import unittest +from . import case import warnings from collections import defaultdict from concurrent.futures import Future, CancelledError, wait @@ -45,8 +46,6 @@ from xmlrpc import client as xmlrpclib import requests import werkzeug.urls -import werkzeug.urls -from decorator import decorator from lxml import etree, html import odoo @@ -62,14 +61,24 @@ from odoo.tools import float_compare, single_email_re, profiler, lower_logging from odoo.tools.misc import find_in_path from odoo.tools.safe_eval import safe_eval +try: + # the behaviour of decorator changed in 5.0.5 changing the structure of the traceback when + # an error is raised inside a method using a decorator. + # this is not a hudge problem for test execution but this makes error message + # more difficult to read and breaks test_with_decorators + # This also changes the error format making runbot error matching fail + # This also breaks the first frame meaning that the module detection will also fail on runbot + # In 5.1 decoratorx was introduced and it looks like it has the same behaviour of old decorator + from decorator import decoratorx as decorator +except ImportError: + from decorator import decorator + try: import websocket except ImportError: # chrome headless tests will be skipped websocket = None -from .runner import stats_logger - _logger = logging.getLogger(__name__) # The odoo library is supposed already configured. @@ -82,7 +91,6 @@ CHECK_BROWSER_SLEEP = 0.1 # seconds CHECK_BROWSER_ITERATIONS = 100 BROWSER_WAIT = CHECK_BROWSER_SLEEP * CHECK_BROWSER_ITERATIONS # seconds - def get_db_name(): db = odoo.tools.config['db_name'] # If the database name is not provided on the command-line, @@ -188,141 +196,6 @@ class RecordCapturer: return self._model.search(self._domain, order='id') - self._before return self._after -# ------------------------------------------------------------ -# Main classes -# ------------------------------------------------------------ -if sys.version_info >= (3, 8): - BackportSuite = unittest.suite.TestSuite -else: - class BackportSuite(unittest.suite.TestSuite): - # Partial backport of bpo-24412, merged in CPython 3.8 - - def _handleClassSetUp(self, test, result): - previousClass = getattr(result, '_previousTestClass', None) - currentClass = test.__class__ - if currentClass == previousClass: - return - if result._moduleSetUpFailed: - return - if getattr(currentClass, "__unittest_skip__", False): - return - - try: - currentClass._classSetupFailed = False - except TypeError: - # test may actually be a function - # so its class will be a builtin-type - pass - - setUpClass = getattr(currentClass, 'setUpClass', None) - if setUpClass is not None: - unittest.suite._call_if_exists(result, '_setupStdout') - try: - setUpClass() - except Exception as e: - if isinstance(result, unittest.suite._DebugResult): - raise - currentClass._classSetupFailed = True - className = unittest.util.strclass(currentClass) - self._createClassOrModuleLevelException(result, e, - 'setUpClass', - className) - finally: - unittest.suite._call_if_exists(result, '_restoreStdout') - if currentClass._classSetupFailed is True: - if hasattr(currentClass, 'doClassCleanups'): - currentClass.doClassCleanups() - if len(currentClass.tearDown_exceptions) > 0: - for exc in currentClass.tearDown_exceptions: - self._createClassOrModuleLevelException( - result, exc[1], 'setUpClass', className, - info=exc) - - def _createClassOrModuleLevelException(self, result, exc, method_name, parent, info=None): - errorName = f'{method_name} ({parent})' - self._addClassOrModuleLevelException(result, exc, errorName, info) - - def _addClassOrModuleLevelException(self, result, exception, errorName, info=None): - error = unittest.suite._ErrorHolder(errorName) - addSkip = getattr(result, 'addSkip', None) - if addSkip is not None and isinstance(exception, unittest.case.SkipTest): - addSkip(error, str(exception)) - else: - if not info: - result.addError(error, sys.exc_info()) - else: - result.addError(error, info) - - def _tearDownPreviousClass(self, test, result): - previousClass = getattr(result, '_previousTestClass', None) - currentClass = test.__class__ - if currentClass == previousClass: - return - if getattr(previousClass, '_classSetupFailed', False): - return - if getattr(result, '_moduleSetUpFailed', False): - return - if getattr(previousClass, "__unittest_skip__", False): - return - - tearDownClass = getattr(previousClass, 'tearDownClass', None) - if tearDownClass is not None: - unittest.suite._call_if_exists(result, '_setupStdout') - try: - tearDownClass() - except Exception as e: - if isinstance(result, unittest.suite._DebugResult): - raise - className = unittest.util.strclass(previousClass) - self._createClassOrModuleLevelException(result, e, - 'tearDownClass', - className) - finally: - unittest.suite._call_if_exists(result, '_restoreStdout') - if hasattr(previousClass, 'doClassCleanups'): - previousClass.doClassCleanups() - if len(previousClass.tearDown_exceptions) > 0: - for exc in previousClass.tearDown_exceptions: - className = unittest.util.strclass(previousClass) - self._createClassOrModuleLevelException(result, exc[1], - 'tearDownClass', - className, - info=exc) - -class OdooSuite(BackportSuite): - def _handleClassSetUp(self, test, result): - previous_test_class = getattr(result, '_previousTestClass', None) - if not ( - previous_test_class != type(test) - and hasattr(result, 'stats') - and stats_logger.isEnabledFor(logging.INFO) - ): - super()._handleClassSetUp(test, result) - return - - test_class = type(test) - test_id = f'{test_class.__module__}.{test_class.__qualname__}.setUpClass' - with result.collectStats(test_id): - super()._handleClassSetUp(test, result) - - def _tearDownPreviousClass(self, test, result): - previous_test_class = getattr(result, '_previousTestClass', None) - if not ( - previous_test_class - and previous_test_class != type(test) - and hasattr(result, 'stats') - and stats_logger.isEnabledFor(logging.INFO) - ): - super()._tearDownPreviousClass(test, result) - return - - test_id = f'{previous_test_class.__module__}.{previous_test_class.__qualname__}.tearDownClass' - with result.collectStats(test_id): - super()._tearDownPreviousClass(test, result) - - def has_http_case(self): - return self.countTestCases() and any(isinstance(test_case, HttpCase) for test_case in self) - class MetaCase(type): """ Metaclass of test case classes to assign default 'test_tags': @@ -359,36 +232,14 @@ def _normalize_arch_for_assert(arch_string, parser_method="xml"): return etree.tostring(arch_string, pretty_print=True, encoding='unicode') -class BaseCase(unittest.TestCase, metaclass=MetaCase): +class BaseCase(case.TestCase, metaclass=MetaCase): """ Subclass of TestCase for Odoo-specific code. This class is abstract and expects self.registry, self.cr and self.uid to be initialized by subclasses. """ - _python_version = sys.version_info - if _python_version < (3, 8): - # Partial backport of bpo-24412, merged in CPython 3.8 - _class_cleanups = [] - - @classmethod - def addClassCleanup(cls, function, *args, **kwargs): - """Same as addCleanup, except the cleanup items are called even if - setUpClass fails (unlike tearDownClass). Backport of bpo-24412.""" - cls._class_cleanups.append((function, args, kwargs)) - - @classmethod - def doClassCleanups(cls): - """Execute all class cleanup functions. Normally called for you after tearDownClass. - Backport of bpo-24412.""" - cls.tearDown_exceptions = [] - while cls._class_cleanups: - function, args, kwargs = cls._class_cleanups.pop() - try: - function(*args, **kwargs) - except Exception as exc: - cls.tearDown_exceptions.append(sys.exc_info()) - longMessage = True # more verbose error message by default: https://www.odoo.com/r/Vmh warm = True # False during warm-up phase (see :func:`warmup`) + _python_version = sys.version_info def __init__(self, methodName='runTest'): super().__init__(methodName) @@ -419,9 +270,6 @@ class BaseCase(unittest.TestCase, metaclass=MetaCase): if not failure: break - def shortDescription(self): - return None - def cursor(self): return self.registry.cursor() @@ -805,100 +653,6 @@ class BaseCase(unittest.TestCase, metaclass=MetaCase): profile_session=self.profile_session, **kwargs) - def _callSetUp(self): - # This override is aimed at providing better error logs inside tests. - # First, we want errors to be logged whenever they appear instead of - # after the test, as the latter makes debugging harder and can even be - # confusing in the case of subtests. - # - # When a subtest is used inside a test, (1) the recovered traceback is - # not complete, and (2) the error is delayed to the end of the test - # method. There is unfortunately no simple way to hook inside a subtest - # to fix this issue. The method TestCase.subTest uses the context - # manager _Outcome.testPartExecutor as follows: - # - # with self._outcome.testPartExecutor(self._subtest, isTest=True): - # yield - # - # This context manager is actually also used for the setup, test method, - # teardown, cleanups. If an error occurs during any one of those, it is - # simply appended in TestCase._outcome.errors, and the latter is - # consumed at the end calling _feedErrorsToResult. - # - # The TestCase._outcome is set just before calling _callSetUp. This - # method is actually executed inside a testPartExecutor. Replacing it - # here ensures that all errors will be caught. - # See https://github.com/odoo/odoo/pull/107572 for more info. - self._outcome.errors = _ErrorCatcher(self) - super()._callSetUp() - - -class _ErrorCatcher(list): - """ This extends a list where errors are appended whenever they occur. The - purpose of this class is to feed the errors directly to the output, instead - of letting them accumulate until the test is over. It also improves the - traceback to make it easier to debug. - """ - __slots__ = ['test'] - - def __init__(self, test): - super().__init__() - self.test = test - - def append(self, error): - exc_info = error[1] - if exc_info is not None: - exception_type, exception, tb = exc_info - tb = self._complete_traceback(tb) - exc_info = (exception_type, exception, tb) - self.test._feedErrorsToResult(self.test._outcome.result, [(error[0], exc_info)]) - - def _complete_traceback(self, initial_tb): - Traceback = type(initial_tb) - - # make the set of frames in the traceback - tb_frames = set() - tb = initial_tb - while tb: - tb_frames.add(tb.tb_frame) - tb = tb.tb_next - tb = initial_tb - - # find the common frame by searching the last frame of the current_stack present in the traceback. - current_frame = inspect.currentframe() - common_frame = None - while current_frame: - if current_frame in tb_frames: - common_frame = current_frame # we want to find the last frame in common - current_frame = current_frame.f_back - - if not common_frame: # not really useful but safer - _logger.warning('No common frame found with current stack, displaying full stack') - tb = initial_tb - else: - # remove the tb_frames untile the common_frame is reached (keep the current_frame tb since the line is more accurate) - while tb and tb.tb_frame != common_frame: - tb = tb.tb_next - - # add all current frame elements under the common_frame to tb - current_frame = common_frame.f_back - while current_frame: - tb = Traceback(tb, current_frame, current_frame.f_lasti, current_frame.f_lineno) - current_frame = current_frame.f_back - - # remove traceback root part (odoo_bin, main, loading, ...), as - # everything under the testCase is not useful. Using '_callTestMethod', - # '_callSetUp', '_callTearDown', '_callCleanup' instead of the test - # method since the error does not comme especially from the test method. - while tb: - code = tb.tb_frame.f_code - if code.co_filename.endswith('/unittest/case.py') and code.co_name in ('_callTestMethod', '_callSetUp', '_callTearDown', '_callCleanup'): - return tb.tb_next - tb = tb.tb_next - - _logger.warning('No root frame found, displaying full stacks') - return initial_tb # this shouldn't be reached - savepoint_seq = itertools.count() @@ -3081,72 +2835,3 @@ def tagged(*tags): obj.test_tags = (getattr(obj, 'test_tags', set()) | include) - exclude return obj return tags_decorator - - -class TagsSelector(object): - """ Test selector based on tags. """ - filter_spec_re = re.compile(r'^([+-]?)(\*|\w*)(?:/(\w*))?(?::(\w*))?(?:\.(\w*))?$') # [-][tag][/module][:class][.method] - - def __init__(self, spec): - """ Parse the spec to determine tags to include and exclude. """ - filter_specs = {t.strip() for t in spec.split(',') if t.strip()} - self.exclude = set() - self.include = set() - - for filter_spec in filter_specs: - match = self.filter_spec_re.match(filter_spec) - if not match: - _logger.error('Invalid tag %s', filter_spec) - continue - - sign, tag, module, klass, method = match.groups() - is_include = sign != '-' - - if not tag and is_include: - # including /module:class.method implicitly requires 'standard' - tag = 'standard' - elif not tag or tag == '*': - # '*' indicates all tests (instead of 'standard' tests only) - tag = None - test_filter = (tag, module, klass, method) - - if is_include: - self.include.add(test_filter) - else: - self.exclude.add(test_filter) - - if self.exclude and not self.include: - self.include.add(('standard', None, None, None)) - - def check(self, test): - """ Return whether ``arg`` matches the specification: it must have at - least one tag in ``self.include`` and none in ``self.exclude`` for each tag category. - """ - if not hasattr(test, 'test_tags'): # handle the case where the Test does not inherit from BaseCase and has no test_tags - _logger.debug("Skipping test '%s' because no test_tag found.", test) - return False - - test_module = getattr(test, 'test_module', None) - test_class = getattr(test, 'test_class', None) - test_tags = test.test_tags | {test_module} # module as test_tags deprecated, keep for retrocompatibility, - test_method = getattr(test, '_testMethodName', None) - - def _is_matching(test_filter): - (tag, module, klass, method) = test_filter - if tag and tag not in test_tags: - return False - elif module and module != test_module: - return False - elif klass and klass != test_class: - return False - elif method and test_method and method != test_method: - return False - return True - - if any(_is_matching(test_filter) for test_filter in self.exclude): - return False - - if any(_is_matching(test_filter) for test_filter in self.include): - return True - - return False diff --git a/odoo/tests/loader.py b/odoo/tests/loader.py index 3f343001b13..7906a07fb74 100644 --- a/odoo/tests/loader.py +++ b/odoo/tests/loader.py @@ -2,16 +2,15 @@ import importlib import importlib.util import inspect import itertools -import logging import threading import unittest from .. import tools -from .common import TagsSelector, OdooSuite -from .runner import OdooTestResult +from .tag_selector import TagsSelector +from .suite import OdooSuite +from .result import OdooTestResult -_logger = logging.getLogger(__name__) def get_test_modules(module): """ Return a list of module for the addons potentially containing tests to feed unittest.TestLoader.loadTestsFromModule() """ @@ -23,6 +22,7 @@ def get_test_modules(module): return results + def _get_tests_modules(mod): spec = importlib.util.find_spec('.tests', mod.name) if not spec: @@ -35,6 +35,7 @@ def _get_tests_modules(mod): if name.startswith('test_') ] + def make_suite(module_names, position='at_install'): """ Creates a test suite for all the tests in the specified modules, filtered by the provided ``position`` and the current test tags @@ -53,6 +54,7 @@ def make_suite(module_names, position='at_install'): ) return OdooSuite(sorted(tests, key=lambda t: t.test_sequence)) + def run_suite(suite, module_name=None): # avoid dependency hell from ..modules import module @@ -66,6 +68,7 @@ def run_suite(suite, module_name=None): module.current_test = None return results + def unwrap_suite(test): """ Attempts to unpack testsuites (holding suites or cases) in order to @@ -83,10 +86,10 @@ def unwrap_suite(test): return subtests = list(test) - # custom test suite (no test cases) - if not len(subtests): - yield test - return + ## custom test suite (no test cases) + #if not len(subtests): + # yield test + # return for item in itertools.chain.from_iterable(unwrap_suite(t) for t in subtests): yield item diff --git a/odoo/tests/result.py b/odoo/tests/result.py index c7e3206d749..b39dab953d5 100644 --- a/odoo/tests/result.py +++ b/odoo/tests/result.py @@ -1,56 +1,85 @@ """Test result object""" -import io -import sys +import logging +import collections +import contextlib +import inspect +import re +import time import traceback -from . import util -from functools import wraps +from typing import NamedTuple + +from . import case +from .. import sql_db __unittest = True -def failfast(method): - @wraps(method) - def inner(self, *args, **kw): - if getattr(self, 'failfast', False): - self.stop() - return method(self, *args, **kw) - return inner - STDOUT_LINE = '\nStdout:\n%s' STDERR_LINE = '\nStderr:\n%s' -class TestResult(object): - """Holder for test result information. +stats_logger = logging.getLogger('odoo.tests.stats') + + +class Stat(NamedTuple): + time: float = 0.0 + queries: int = 0 + + def __add__(self, other: 'Stat') -> 'Stat': + if other == 0: + return self + + if not isinstance(other, Stat): + return NotImplemented + + return Stat( + self.time + other.time, + self.queries + other.queries, + ) + +_logger = logging.getLogger(__name__) +_TEST_ID = re.compile(r""" +^ +odoo\.addons\. +(?P[^.]+) +\.tests\. +(?P.+) +\. +(?P[^.]+) +$ +""", re.VERBOSE) + + +class OdooTestResult(object): + """ + This class in inspired from TextTestResult and modifies TestResult + Instead of using a stream, we are using the logger. + + unittest.TestResult: Holder for test result information. Test results are automatically managed by the TestCase and TestSuite classes, and do not need to be explicitly manipulated by writers of tests. - Each instance holds the total number of tests run, and collections of - failures and errors that occurred among those test runs. The collections - contain tuples of (testcase, exceptioninfo), where exceptioninfo is the - formatted traceback of the error that occurred. + This version does not hold a list of failure but just a count since the failure is logged immediately + This version is also simplied to better match our use cases """ + _previousTestClass = None - _testRunEntered = False _moduleSetUpFailed = False + def __init__(self, stream=None, descriptions=None, verbosity=None): - self.failfast = False - self.failures = [] - self.errors = [] + self.failures_count = 0 + self.errors_count = 0 self.testsRun = 0 - self.skipped = [] - self.expectedFailures = [] - self.unexpectedSuccesses = [] - self.shouldStop = False - self.buffer = False + self.skipped = 0 self.tb_locals = False - self._stdout_buffer = None - self._stderr_buffer = None - self._original_stdout = sys.stdout - self._original_stderr = sys.stderr - self._mirrorOutput = False + # custom + self.time_start = None + self.queries_start = None + self._soft_fail = False + self.had_failure = False + self.stats = collections.defaultdict(Stat) def printErrors(self): "Called by TestRunner after test run" @@ -58,117 +87,58 @@ class TestResult(object): def startTest(self, test): "Called when the given test is about to be run" self.testsRun += 1 - self._mirrorOutput = False - self._setupStdout() - - def _setupStdout(self): - if self.buffer: - if self._stderr_buffer is None: - self._stderr_buffer = io.StringIO() - self._stdout_buffer = io.StringIO() - sys.stdout = self._stdout_buffer - sys.stderr = self._stderr_buffer - - def startTestRun(self): - """Called once before any tests are executed. - - See startTest for a method called before each test. - """ + self.log(logging.INFO, 'Starting %s ...', self.getDescription(test), test=test) + self.time_start = time.time() + self.queries_start = sql_db.sql_counter def stopTest(self, test): """Called when the given test has been run""" - self._restoreStdout() - self._mirrorOutput = False + if stats_logger.isEnabledFor(logging.INFO): + self.stats[test.id()] = Stat( + time=time.time() - self.time_start, + queries=sql_db.sql_counter - self.queries_start, + ) - def _restoreStdout(self): - if self.buffer: - if self._mirrorOutput: - output = sys.stdout.getvalue() - error = sys.stderr.getvalue() - if output: - if not output.endswith('\n'): - output += '\n' - self._original_stdout.write(STDOUT_LINE % output) - if error: - if not error.endswith('\n'): - error += '\n' - self._original_stderr.write(STDERR_LINE % error) - - sys.stdout = self._original_stdout - sys.stderr = self._original_stderr - self._stdout_buffer.seek(0) - self._stdout_buffer.truncate() - self._stderr_buffer.seek(0) - self._stderr_buffer.truncate() - - def stopTestRun(self): - """Called once after all tests are executed. - - See stopTest for a method called after each test. - """ - - @failfast def addError(self, test, err): """Called when an error has occurred. 'err' is a tuple of values as returned by sys.exc_info(). """ - self.errors.append((test, self._exc_info_to_string(err, test))) - self._mirrorOutput = True + if self._soft_fail: + self.had_failure = True + else: + self.errors_count += 1 + self.logError("ERROR", test, err) - @failfast def addFailure(self, test, err): """Called when an error has occurred. 'err' is a tuple of values as returned by sys.exc_info().""" - self.failures.append((test, self._exc_info_to_string(err, test))) - self._mirrorOutput = True + if self._soft_fail: + self.had_failure = True + else: + self.failures_count += 1 + self.logError("FAIL", test, err) def addSubTest(self, test, subtest, err): - """Called at the end of a subtest. - 'err' is None if the subtest ended successfully, otherwise it's a - tuple of values as returned by sys.exc_info(). - """ - # By default, we don't do anything with successful subtests, but - # more sophisticated test results might want to record them. if err is not None: - if getattr(self, 'failfast', False): - self.stop() if issubclass(err[0], test.failureException): - errors = self.failures + self.addFailure(subtest, err) else: - errors = self.errors - errors.append((subtest, self._exc_info_to_string(err, test))) - self._mirrorOutput = True + self.addError(subtest, err) def addSuccess(self, test): "Called when a test has completed successfully" - pass def addSkip(self, test, reason): """Called when a test is skipped.""" - self.skipped.append((test, reason)) - - def addExpectedFailure(self, test, err): - """Called when an expected failure/error occurred.""" - self.expectedFailures.append( - (test, self._exc_info_to_string(err, test))) - - @failfast - def addUnexpectedSuccess(self, test): - """Called when a test was expected to fail, but succeed.""" - self.unexpectedSuccesses.append(test) + self.skipped += 1 + self.log(logging.INFO, 'skipped %s : %s', self.getDescription(test), reason, test=test) def wasSuccessful(self): """Tells whether or not this result was a success.""" # The hasattr check is for test_result's OldResult test. That # way this method works on objects that lack the attribute. # (where would such result intances come from? old stored pickles?) - return ((len(self.failures) == len(self.errors) == 0) and - (not hasattr(self, 'unexpectedSuccesses') or - len(self.unexpectedSuccesses) == 0)) - - def stop(self): - """Indicates that the tests should be aborted.""" - self.shouldStop = True + return self.failures_count == self.errors_count == 0 def _exc_info_to_string(self, err, test): """Converts a sys.exc_info()-style tuple of values into a string.""" @@ -186,20 +156,8 @@ class TestResult(object): exctype, value, tb, limit=length, capture_locals=self.tb_locals) msgLines = list(tb_e.format()) - if self.buffer: - output = sys.stdout.getvalue() - error = sys.stderr.getvalue() - if output: - if not output.endswith('\n'): - output += '\n' - msgLines.append(STDOUT_LINE % output) - if error: - if not error.endswith('\n'): - error += '\n' - msgLines.append(STDERR_LINE % error) return ''.join(msgLines) - def _is_relevant_tb_level(self, tb): return '__unittest' in tb.tb_frame.f_globals @@ -211,6 +169,156 @@ class TestResult(object): return length def __repr__(self): - return ("<%s run=%i errors=%i failures=%i>" % - (util.strclass(self.__class__), self.testsRun, len(self.errors), - len(self.failures))) + return ("<%s.%s run=%i errors=%i failures=%i>" % + (self.__class__.__module__, self.__class__.__qualname__, self.testsRun, len(self.errors_count), len(self.failures_count))) + + def __str__(self): + return f'{self.failures_count} failed, {self.errors_count} error(s) of {self.testsRun} tests' + + + @contextlib.contextmanager + def soft_fail(self): + self.had_failure = False + self._soft_fail = True + try: + yield + finally: + self._soft_fail = False + self.had_failure = False + + def update(self, other): + """ Merges an other test result into this one, only updates contents + + :type other: OdooTestResult + """ + self.failures_count += other.failures_count + self.errors_count += other.errors_count + self.testsRun += other.testsRun + self.skipped += other.skipped + self.stats.update(other.stats) + + def log(self, level, msg, *args, test=None, exc_info=None, extra=None, stack_info=False, caller_infos=None): + """ + ``test`` is the running test case, ``caller_infos`` is + (fn, lno, func, sinfo) (logger.findCaller format), see logger.log for + the other parameters. + """ + test = test or self + while isinstance(test, case._SubTest) and test.test_case: + test = test.test_case + logger = logging.getLogger(test.__module__) + try: + caller_infos = caller_infos or logger.findCaller(stack_info) + except ValueError: + caller_infos = "(unknown file)", 0, "(unknown function)", None + (fn, lno, func, sinfo) = caller_infos + # using logger.log makes it difficult to spot-replace findCaller in + # order to provide useful location information (the problematic spot + # inside the test function), so use lower-level functions instead + if logger.isEnabledFor(level): + record = logger.makeRecord(logger.name, level, fn, lno, msg, args, exc_info, func, extra, sinfo) + logger.handle(record) + + def log_stats(self): + if not stats_logger.isEnabledFor(logging.INFO): + return + + details = stats_logger.isEnabledFor(logging.DEBUG) + stats_tree = collections.defaultdict(Stat) + counts = collections.Counter() + for test, stat in self.stats.items(): + r = _TEST_ID.match(test) + if not r: # upgrade has tests at weird paths, ignore them + continue + + stats_tree[r['module']] += stat + counts[r['module']] += 1 + if details: + stats_tree['%(module)s.%(class)s' % r] += stat + stats_tree['%(module)s.%(class)s.%(method)s' % r] += stat + + if details: + stats_logger.debug('Detailed Tests Report:\n%s', ''.join( + f'\t{test}: {stats.time:.2f}s {stats.queries} queries\n' + for test, stats in sorted(stats_tree.items()) + )) + else: + for module, stat in sorted(stats_tree.items()): + stats_logger.info( + "%s: %d tests %.2fs %d queries", + module, counts[module], + stat.time, stat.queries + ) + + def getDescription(self, test): + if isinstance(test, case._SubTest): + return 'Subtest %s.%s %s' % (test.test_case.__class__.__qualname__, test.test_case._testMethodName, test._subDescription()) + if isinstance(test, case.TestCase): + # since we have the module name in the logger, this will avoid to duplicate module info in log line + # we only apply this for TestCase since we can receive error handler or other special case + return "%s.%s" % (test.__class__.__qualname__, test._testMethodName) + return str(test) + + @contextlib.contextmanager + def collectStats(self, test_id): + queries_before = sql_db.sql_counter + time_start = time.time() + + yield + + self.stats[test_id] += Stat( + time=time.time() - time_start, + queries=sql_db.sql_counter - queries_before, + ) + + def logError(self, flavour, test, error): + err = self._exc_info_to_string(error, test) + caller_infos = self.getErrorCallerInfo(error, test) + self.log(logging.INFO, '=' * 70, test=test, caller_infos=caller_infos) # keep this as info !!!!!! + self.log(logging.ERROR, "%s: %s\n%s", flavour, self.getDescription(test), err, test=test, caller_infos=caller_infos) + + def getErrorCallerInfo(self, error, test): + """ + :param error: A tuple (exctype, value, tb) as returned by sys.exc_info(). + :param test: A TestCase that created this error. + :returns: a tuple (fn, lno, func, sinfo) matching the logger findCaller format or None + """ + + # only handle TestCase here. test can be an _ErrorHolder in some case (setup/teardown class errors) + if not isinstance(test, case.TestCase): + return + + _, _, error_traceback = error + + # move upwards the subtest hierarchy to find the real test + while isinstance(test, case._SubTest) and test.test_case: + test = test.test_case + + method_tb = None + file_tb = None + filename = inspect.getfile(type(test)) + + # Note: since _ErrorCatcher was introduced, we could always take the + # last frame, keeping the check on the test method for safety. + # Fallbacking on file for cleanup file shoud always be correct to a + # minimal working version would be + # + # infos_tb = error_traceback + # while infos_tb.tb_next() + # infos_tb = infos_tb.tb_next() + # + while error_traceback: + code = error_traceback.tb_frame.f_code + if code.co_name in (test._testMethodName, 'setUp', 'tearDown'): + method_tb = error_traceback + if code.co_filename == filename: + file_tb = error_traceback + error_traceback = error_traceback.tb_next + + infos_tb = method_tb or file_tb + if infos_tb: + code = infos_tb.tb_frame.f_code + lineno = infos_tb.tb_lineno + filename = code.co_filename + method = test._testMethodName + return (filename, lineno, method, None) diff --git a/odoo/tests/runner.py b/odoo/tests/runner.py deleted file mode 100644 index c16f728c635..00000000000 --- a/odoo/tests/runner.py +++ /dev/null @@ -1,261 +0,0 @@ -import collections -import contextlib -import inspect -import logging -import re -import time -import unittest -from typing import NamedTuple - -from .. import sql_db - -stats_logger = logging.getLogger('odoo.tests.stats') - -class Stat(NamedTuple): - time: float = 0.0 - queries: int = 0 - - def __add__(self, other: 'Stat') -> 'Stat': - if other == 0: - return self - - if not isinstance(other, Stat): - return NotImplemented - - return Stat( - self.time + other.time, - self.queries + other.queries, - ) - -_logger = logging.getLogger(__name__) -_TEST_ID = re.compile(r""" -^ -odoo\.addons\. -(?P[^.]+) -\.tests\. -(?P.+) -\. -(?P[^.]+) -$ -""", re.VERBOSE) -class OdooTestResult(unittest.result.TestResult): - """ - This class in inspired from TextTestResult (https://github.com/python/cpython/blob/master/Lib/unittest/runner.py) - Instead of using a stream, we are using the logger, - but replacing the "findCaller" in order to give the information we - have based on the test object that is running. - """ - - def __init__(self): - super().__init__() - self.time_start = None - self.queries_start = None - self._soft_fail = False - self.had_failure = False - self.stats = collections.defaultdict(Stat) - - def __str__(self): - return f'{len(self.failures)} failed, {len(self.errors)} error(s) of {self.testsRun} tests' - - @contextlib.contextmanager - def soft_fail(self): - self.had_failure = False - self._soft_fail = True - try: - yield - finally: - self._soft_fail = False - self.had_failure = False - - def update(self, other): - """ Merges an other test result into this one, only updates contents - - :type other: OdooTestResult - """ - self.failures.extend(other.failures) - self.errors.extend(other.errors) - self.testsRun += other.testsRun - self.skipped.extend(other.skipped) - self.expectedFailures.extend(other.expectedFailures) - self.unexpectedSuccesses.extend(other.unexpectedSuccesses) - self.shouldStop = self.shouldStop or other.shouldStop - self.stats.update(other.stats) - - def log(self, level, msg, *args, test=None, exc_info=None, extra=None, stack_info=False, caller_infos=None): - """ - ``test`` is the running test case, ``caller_infos`` is - (fn, lno, func, sinfo) (logger.findCaller format), see logger.log for - the other parameters. - """ - test = test or self - while isinstance(test, unittest.case._SubTest) and test.test_case: - test = test.test_case - logger = logging.getLogger(test.__module__) - try: - caller_infos = caller_infos or logger.findCaller(stack_info) - except ValueError: - caller_infos = "(unknown file)", 0, "(unknown function)", None - (fn, lno, func, sinfo) = caller_infos - # using logger.log makes it difficult to spot-replace findCaller in - # order to provide useful location information (the problematic spot - # inside the test function), so use lower-level functions instead - if logger.isEnabledFor(level): - record = logger.makeRecord(logger.name, level, fn, lno, msg, args, exc_info, func, extra, sinfo) - logger.handle(record) - - def log_stats(self): - if not stats_logger.isEnabledFor(logging.INFO): - return - - details = stats_logger.isEnabledFor(logging.DEBUG) - stats_tree = collections.defaultdict(Stat) - counts = collections.Counter() - for test, stat in self.stats.items(): - r = _TEST_ID.match(test) - if not r: # upgrade has tests at weird paths, ignore them - continue - - stats_tree[r['module']] += stat - counts[r['module']] += 1 - if details: - stats_tree['%(module)s.%(class)s' % r] += stat - stats_tree['%(module)s.%(class)s.%(method)s' % r] += stat - - if details: - stats_logger.debug('Detailed Tests Report:\n%s', ''.join( - f'\t{test}: {stats.time:.2f}s {stats.queries} queries\n' - for test, stats in sorted(stats_tree.items()) - )) - else: - for module, stat in sorted(stats_tree.items()): - stats_logger.info( - "%s: %d tests %.2fs %d queries", - module, counts[module], - stat.time, stat.queries - ) - - def getDescription(self, test): - if isinstance(test, unittest.case._SubTest): - return 'Subtest %s.%s %s' % (test.test_case.__class__.__qualname__, test.test_case._testMethodName, test._subDescription()) - if isinstance(test, unittest.TestCase): - # since we have the module name in the logger, this will avoid to duplicate module info in log line - # we only apply this for TestCase since we can receive error handler or other special case - return "%s.%s" % (test.__class__.__qualname__, test._testMethodName) - return str(test) - - def startTest(self, test): - super().startTest(test) - self.log(logging.INFO, 'Starting %s ...', self.getDescription(test), test=test) - self.time_start = time.time() - self.queries_start = sql_db.sql_counter - - def stopTest(self, test): - if stats_logger.isEnabledFor(logging.INFO): - self.stats[test.id()] = Stat( - time=time.time() - self.time_start, - queries=sql_db.sql_counter - self.queries_start, - ) - super().stopTest(test) - - @contextlib.contextmanager - def collectStats(self, test_id): - queries_before = sql_db.sql_counter - time_start = time.time() - - yield - - self.stats[test_id] += Stat( - time=time.time() - time_start, - queries=sql_db.sql_counter - queries_before, - ) - - def addError(self, test, err): - if self._soft_fail: - self.had_failure = True - else: - super().addError(test, err) - self.logError("ERROR", test, err) - - def addFailure(self, test, err): - if self._soft_fail: - self.had_failure = True - else: - super().addFailure(test, err) - self.logError("FAIL", test, err) - - def addSubTest(self, test, subtest, err): - # since addSubTest is not making a call to addFailure or addError we need to manage it too - # https://github.com/python/cpython/blob/3.7/Lib/unittest/result.py#L136 - if err is not None: - if issubclass(err[0], test.failureException): - flavour = "FAIL" - else: - flavour = "ERROR" - self.logError(flavour, subtest, err) - if self._soft_fail: - self.had_failure = True - err = None - super().addSubTest(test, subtest, err) - - def addSkip(self, test, reason): - super().addSkip(test, reason) - self.log(logging.INFO, 'skipped %s : %s', self.getDescription(test), reason, test=test) - - def addUnexpectedSuccess(self, test): - super().addUnexpectedSuccess(test) - self.log(logging.ERROR, 'unexpected success for %s', self.getDescription(test), test=test) - - def logError(self, flavour, test, error): - err = self._exc_info_to_string(error, test) - caller_infos = self.getErrorCallerInfo(error, test) - self.log(logging.INFO, '=' * 70, test=test, caller_infos=caller_infos) # keep this as info !!!!!! - self.log(logging.ERROR, "%s: %s\n%s", flavour, self.getDescription(test), err, test=test, caller_infos=caller_infos) - - def getErrorCallerInfo(self, error, test): - """ - :param error: A tuple (exctype, value, tb) as returned by sys.exc_info(). - :param test: A TestCase that created this error. - :returns: a tuple (fn, lno, func, sinfo) matching the logger findCaller format or None - """ - - # only test case should be executed in odoo, this is only a safe guard - if isinstance(test, unittest.suite._ErrorHolder): - return - if not isinstance(test, unittest.TestCase): - _logger.warning('%r is not a TestCase' % test) - return - - _, _, error_traceback = error - - # move upwards the subtest hierarchy to find the real test - while isinstance(test, unittest.case._SubTest) and test.test_case: - test = test.test_case - - method_tb = None - file_tb = None - filename = inspect.getfile(type(test)) - - # Note: since _ErrorCatcher was introduced, we could always take the - # last frame, keeping the check on the test method for safety. - # Fallbacking on file for cleanup file shoud always be correct to a - # minimal working version would be - # - # infos_tb = error_traceback - # while infos_tb.tb_next() - # infos_tb = infos_tb.tb_next() - # - while error_traceback: - code = error_traceback.tb_frame.f_code - if code.co_name in (test._testMethodName, 'setUp', 'tearDown'): - method_tb = error_traceback - if code.co_filename == filename: - file_tb = error_traceback - error_traceback = error_traceback.tb_next - - infos_tb = method_tb or file_tb - if infos_tb: - code = infos_tb.tb_frame.f_code - lineno = infos_tb.tb_lineno - filename = code.co_filename - method = test._testMethodName - return (filename, lineno, method, None) diff --git a/odoo/tests/suite.py b/odoo/tests/suite.py index 41993f9cf69..4a64ec01af5 100644 --- a/odoo/tests/suite.py +++ b/odoo/tests/suite.py @@ -1,97 +1,31 @@ -"""TestSuite""" +""" +Vendor unittest.TestSuite +This is a modified version of python 3.8 unitest.TestSuite + +Odoo tests customisation combined with the need of a cross version compatibility +started to make TestSuite and other unitest object more complicated than vendoring +the part we need for Odoo. This versions is simplified in order +to minimise the code to maintain + +- Removes expected failure support +- Removes module setUp/tearDown support + +""" + +import logging import sys from . import case -from . import util +from .common import HttpCase +from .result import stats_logger +from unittest import util, BaseTestSuite, TestCase __unittest = True -def _call_if_exists(parent, attr): - func = getattr(parent, attr, lambda: None) - func() - - -class BaseTestSuite(object): - """A simple test suite that doesn't provide class or module shared fixtures. - """ - _cleanup = True - - def __init__(self, tests=()): - self._tests = [] - self._removed_tests = 0 - self.addTests(tests) - - def __repr__(self): - return "<%s tests=%s>" % (util.strclass(self.__class__), list(self)) - - def __eq__(self, other): - if not isinstance(other, self.__class__): - return NotImplemented - return list(self) == list(other) - - def __iter__(self): - return iter(self._tests) - - def countTestCases(self): - cases = self._removed_tests - for test in self: - if test: - cases += test.countTestCases() - return cases - - def addTest(self, test): - # sanity checks - if not callable(test): - raise TypeError("{} is not callable".format(repr(test))) - if isinstance(test, type) and issubclass(test, - (case.TestCase, TestSuite)): - raise TypeError("TestCases and TestSuites must be instantiated " - "before passing them to addTest()") - self._tests.append(test) - - def addTests(self, tests): - if isinstance(tests, str): - raise TypeError("tests must be an iterable of tests, not a string") - for test in tests: - self.addTest(test) - - def run(self, result): - for index, test in enumerate(self): - if result.shouldStop: - break - test(result) - if self._cleanup: - self._removeTestAtIndex(index) - return result - - def _removeTestAtIndex(self, index): - """Stop holding a reference to the TestCase at index.""" - try: - test = self._tests[index] - except TypeError: - # support for suite implementations that have overridden self._tests - pass - else: - # Some unittest tests add non TestCase/TestSuite objects to - # the suite. - if hasattr(test, 'countTestCases'): - self._removed_tests += test.countTestCases() - self._tests[index] = None - - def __call__(self, *args, **kwds): - return self.run(*args, **kwds) - - def debug(self): - """Run the tests without collecting errors in a TestResult""" - for test in self: - test.debug() - - class TestSuite(BaseTestSuite): """A test suite is a composite test consisting of a number of TestCases. - For use, create an instance of TestSuite, then add test case instances. When all tests have been added, the suite can be passed to a test runner, such as TextTestRunner. It will run the individual test cases @@ -100,211 +34,86 @@ class TestSuite(BaseTestSuite): """ def run(self, result, debug=False): - topLevel = False - if getattr(result, '_testRunEntered', False) is False: - result._testRunEntered = topLevel = True + for test in self: + assert isinstance(test, (TestCase)) + self._tearDownPreviousClass(test, result) + self._handleClassSetUp(test, result) + result._previousTestClass = test.__class__ - for index, test in enumerate(self): - if result.shouldStop: - break - - if _isnotsuite(test): - self._tearDownPreviousClass(test, result) - self._handleModuleFixture(test, result) - self._handleClassSetUp(test, result) - result._previousTestClass = test.__class__ - - if (getattr(test.__class__, '_classSetupFailed', False) or - getattr(result, '_moduleSetUpFailed', False)): - continue - - if not debug: + if not test.__class__._classSetupFailed: test(result) - else: - test.debug() - if self._cleanup: - self._removeTestAtIndex(index) - - if topLevel: - self._tearDownPreviousClass(None, result) - self._handleModuleTearDown(result) - result._testRunEntered = False + self._tearDownPreviousClass(None, result) return result - def debug(self): - """Run the tests without collecting errors in a TestResult""" - debug = _DebugResult() - self.run(debug, True) - - ################################ - def _handleClassSetUp(self, test, result): - previousClass = getattr(result, '_previousTestClass', None) + previousClass = result._previousTestClass currentClass = test.__class__ if currentClass == previousClass: return if result._moduleSetUpFailed: return - if getattr(currentClass, "__unittest_skip__", False): + if currentClass.__unittest_skip__: return + currentClass._classSetupFailed = False + try: - currentClass._classSetupFailed = False - except TypeError: - # test may actually be a function - # so its class will be a builtin-type - pass + currentClass.setUpClass() + except Exception as e: + currentClass._classSetupFailed = True + className = util.strclass(currentClass) + self._createClassOrModuleLevelException(result, e, + 'setUpClass', + className) + finally: + if currentClass._classSetupFailed is True: + currentClass.doClassCleanups() + if len(currentClass.tearDown_exceptions) > 0: + for exc in currentClass.tearDown_exceptions: + self._createClassOrModuleLevelException( + result, exc[1], 'setUpClass', className, + info=exc) - setUpClass = getattr(currentClass, 'setUpClass', None) - if setUpClass is not None: - _call_if_exists(result, '_setupStdout') - try: - setUpClass() - except Exception as e: - if isinstance(result, _DebugResult): - raise - currentClass._classSetupFailed = True - className = util.strclass(currentClass) - self._createClassOrModuleLevelException(result, e, - 'setUpClass', - className) - finally: - _call_if_exists(result, '_restoreStdout') - if currentClass._classSetupFailed is True: - currentClass.doClassCleanups() - if len(currentClass.tearDown_exceptions) > 0: - for exc in currentClass.tearDown_exceptions: - self._createClassOrModuleLevelException( - result, exc[1], 'setUpClass', className, - info=exc) - - def _get_previous_module(self, result): - previousModule = None - previousClass = getattr(result, '_previousTestClass', None) - if previousClass is not None: - previousModule = previousClass.__module__ - return previousModule - - - def _handleModuleFixture(self, test, result): - previousModule = self._get_previous_module(result) - currentModule = test.__class__.__module__ - if currentModule == previousModule: - return - - self._handleModuleTearDown(result) - - - result._moduleSetUpFailed = False - try: - module = sys.modules[currentModule] - except KeyError: - return - setUpModule = getattr(module, 'setUpModule', None) - if setUpModule is not None: - _call_if_exists(result, '_setupStdout') - try: - setUpModule() - except Exception as e: - try: - case.doModuleCleanups() - except Exception as exc: - self._createClassOrModuleLevelException(result, exc, - 'setUpModule', - currentModule) - if isinstance(result, _DebugResult): - raise - result._moduleSetUpFailed = True - self._createClassOrModuleLevelException(result, e, - 'setUpModule', - currentModule) - finally: - _call_if_exists(result, '_restoreStdout') - - def _createClassOrModuleLevelException(self, result, exc, method_name, + def _createClassOrModuleLevelException(self, result, exception, method_name, parent, info=None): errorName = f'{method_name} ({parent})' - self._addClassOrModuleLevelException(result, exc, errorName, info) - - def _addClassOrModuleLevelException(self, result, exception, errorName, - info=None): error = _ErrorHolder(errorName) - addSkip = getattr(result, 'addSkip', None) - if addSkip is not None and isinstance(exception, case.SkipTest): - addSkip(error, str(exception)) + if isinstance(exception, case.SkipTest): + result.addSkip(error, str(exception)) else: if not info: result.addError(error, sys.exc_info()) else: result.addError(error, info) - def _handleModuleTearDown(self, result): - previousModule = self._get_previous_module(result) - if previousModule is None: - return - if result._moduleSetUpFailed: - return - - try: - module = sys.modules[previousModule] - except KeyError: - return - - tearDownModule = getattr(module, 'tearDownModule', None) - if tearDownModule is not None: - _call_if_exists(result, '_setupStdout') - try: - tearDownModule() - except Exception as e: - if isinstance(result, _DebugResult): - raise - self._createClassOrModuleLevelException(result, e, - 'tearDownModule', - previousModule) - finally: - _call_if_exists(result, '_restoreStdout') - try: - case.doModuleCleanups() - except Exception as e: - self._createClassOrModuleLevelException(result, e, - 'tearDownModule', - previousModule) - def _tearDownPreviousClass(self, test, result): - previousClass = getattr(result, '_previousTestClass', None) + previousClass = result._previousTestClass currentClass = test.__class__ if currentClass == previousClass: return - if getattr(previousClass, '_classSetupFailed', False): + if not previousClass: return - if getattr(result, '_moduleSetUpFailed', False): + if previousClass._classSetupFailed: return - if getattr(previousClass, "__unittest_skip__", False): + if previousClass.__unittest_skip__: return - - tearDownClass = getattr(previousClass, 'tearDownClass', None) - if tearDownClass is not None: - _call_if_exists(result, '_setupStdout') - try: - tearDownClass() - except Exception as e: - if isinstance(result, _DebugResult): - raise - className = util.strclass(previousClass) - self._createClassOrModuleLevelException(result, e, - 'tearDownClass', - className) - finally: - _call_if_exists(result, '_restoreStdout') - previousClass.doClassCleanups() - if len(previousClass.tearDown_exceptions) > 0: - for exc in previousClass.tearDown_exceptions: - className = util.strclass(previousClass) - self._createClassOrModuleLevelException(result, exc[1], - 'tearDownClass', - className, - info=exc) + try: + previousClass.tearDownClass() + except Exception as e: + className = util.strclass(previousClass) + self._createClassOrModuleLevelException(result, e, + 'tearDownClass', + className) + finally: + previousClass.doClassCleanups() + if len(previousClass.tearDown_exceptions) > 0: + for exc in previousClass.tearDown_exceptions: + className = util.strclass(previousClass) + self._createClassOrModuleLevelException(result, exc[1], + 'tearDownClass', + className, + info=exc) class _ErrorHolder(object): @@ -345,17 +154,37 @@ class _ErrorHolder(object): def countTestCases(self): return 0 -def _isnotsuite(test): - "A crude way to tell apart testcases and suites with duck-typing" - try: - iter(test) - except TypeError: - return True - return False +class OdooSuite(TestSuite): + def _handleClassSetUp(self, test, result): + previous_test_class = result._previousTestClass + if not ( + previous_test_class != type(test) + and hasattr(result, 'stats') + and stats_logger.isEnabledFor(logging.INFO) + ): + super()._handleClassSetUp(test, result) + return -class _DebugResult(object): - "Used by the TestSuite to hold previous class when running in debug." - _previousTestClass = None - _moduleSetUpFailed = False - shouldStop = False + test_class = type(test) + test_id = f'{test_class.__module__}.{test_class.__qualname__}.setUpClass' + with result.collectStats(test_id): + super()._handleClassSetUp(test, result) + + def _tearDownPreviousClass(self, test, result): + previous_test_class = result._previousTestClass + if not ( + previous_test_class + and previous_test_class != type(test) + and hasattr(result, 'stats') + and stats_logger.isEnabledFor(logging.INFO) + ): + super()._tearDownPreviousClass(test, result) + return + + test_id = f'{previous_test_class.__module__}.{previous_test_class.__qualname__}.tearDownClass' + with result.collectStats(test_id): + super()._tearDownPreviousClass(test, result) + + def has_http_case(self): + return self.countTestCases() and any(isinstance(test_case, HttpCase) for test_case in self) diff --git a/odoo/tests/tag_selector.py b/odoo/tests/tag_selector.py new file mode 100644 index 00000000000..8ea9482fcb1 --- /dev/null +++ b/odoo/tests/tag_selector.py @@ -0,0 +1,73 @@ +import re +import logging + +_logger = logging.getLogger(__name__) + + +class TagsSelector(object): + """ Test selector based on tags. """ + filter_spec_re = re.compile(r'^([+-]?)(\*|\w*)(?:/(\w*))?(?::(\w*))?(?:\.(\w*))?$') # [-][tag][/module][:class][.method] + + def __init__(self, spec): + """ Parse the spec to determine tags to include and exclude. """ + filter_specs = {t.strip() for t in spec.split(',') if t.strip()} + self.exclude = set() + self.include = set() + + for filter_spec in filter_specs: + match = self.filter_spec_re.match(filter_spec) + if not match: + _logger.error('Invalid tag %s', filter_spec) + continue + + sign, tag, module, klass, method = match.groups() + is_include = sign != '-' + + if not tag and is_include: + # including /module:class.method implicitly requires 'standard' + tag = 'standard' + elif not tag or tag == '*': + # '*' indicates all tests (instead of 'standard' tests only) + tag = None + test_filter = (tag, module, klass, method) + + if is_include: + self.include.add(test_filter) + else: + self.exclude.add(test_filter) + + if self.exclude and not self.include: + self.include.add(('standard', None, None, None, None)) + + def check(self, test): + """ Return whether ``arg`` matches the specification: it must have at + least one tag in ``self.include`` and none in ``self.exclude`` for each tag category. + """ + if not hasattr(test, 'test_tags'): # handle the case where the Test does not inherit from BaseCase and has no test_tags + _logger.debug("Skipping test '%s' because no test_tag found.", test) + return False + + test_module = test.test_module + test_class = test.test_class + test_tags = test.test_tags | {test_module} # module as test_tags deprecated, keep for retrocompatibility, + test_method = test._testMethodName + + def _is_matching(test_filter): + (tag, module, klass, method) = test_filter + if tag and tag not in test_tags: + return False + elif module and module != test_module: + return False + elif klass and klass != test_class: + return False + elif method and test_method and method != test_method: + return False + return True + + if any(_is_matching(test_filter) for test_filter in self.exclude): + return False + + if any(_is_matching(test_filter) for test_filter in self.include): + return True + + return False