[IMP] tests: refactor unittest classes

Odoo Test environments requires to modify many parts of the unittest
TestCase, Suite and Result.

The main initial reason is to **avoid to postpone result at the end of
the test suite**, because even if it is convenient to have all errors
visible after the tests in some case, odoo logs adds information during
the execution that can be useful to debug when a test fail, to have
context for an error. (see **OdooTestResult**)

We are also fixing the stack trace comming from a unittest and since
there is no proper way to hook inside the TestPartExecutor, a dirty hack
injects anoter result on the outcome to manage the error and complete
the stack trace. This was also a way to avoid to postpone subtest logs
at the end of the test case (see _ErrorCatcher)

`_feedErrorsToResult` was used to test the test suite behavior since
there are many customization and this is quite fragile, especially if
unittest changes behavior in other python version.

**Python 3.11** introduced python/cpython#664448d8 That, in a way, goes
in the same direction of the changed introduced with _ErrorCatcher:
immediately feed errors to resut instead of postponing it. But this also
removes `_feedErrorsToResult` that was used to test this behaviors, as
well as other ones.

Since odoo should remain multi-version, this amount of changes on the
initial behavior become to complicate to keep cross-version and the
(already in our mind for a while) solution to **vendor unittest** will
help to simplify most of our test code base.

This commit modified the vendored unittest files to simplify them as
much as possible to suite our needs.

Since the runner is still the unittest one, we need to inherit from
unittest.Testcase in order to have the right type.

This also means that we still have access to all TestCase methods
without overriding them all. This is convenient for assertion methods as
an example but the initial idea is to vendor our own version of TestCase
to avoid having trouble to adapte our miscommunications to future python
versions. A trade-off must be done to chose what should remain in our
code base. The idea is to keep logic closely linked to our changes in
our code base, mainly around the run method, but also addClassCleanup
wich need to be vendored for python 3.7, but assertions methods are
independent. Any logic can be moved fom unittest to our
vendored version in the future if needed.

X-original-commit: 9a5d1ea54be49e4cc8208c33e76a6bbd2414d5d0
Part-of: odoo/odoo#113850
This commit is contained in:
Xavier-Do
2023-02-28 23:49:33 +01:00
parent 630b6484a2
commit e9b170da38
15 changed files with 661 additions and 2377 deletions
+7
View File
@@ -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
+112 -1353
View File
File diff suppressed because it is too large Load Diff
+15 -330
View File
@@ -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
+11 -8
View File
@@ -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
+237 -129
View File
@@ -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<module>[^.]+)
\.tests\.
(?P<class>.+)
\.
(?P<method>[^.]+)
$
""", 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)
-261
View File
@@ -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<module>[^.]+)
\.tests\.
(?P<class>.+)
\.
(?P<method>[^.]+)
$
""", 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)
+99 -270
View File
@@ -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)
+73
View File
@@ -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