Coverage for adhoc-cicd-odoo-odoo / odoo / tests / result.py: 49%
196 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:05 +0000
1"""Test result object"""
3import collections
4import contextlib
5import inspect
6import logging
7import os
8import re
9import sys
10import time
11import traceback
13from typing import NamedTuple
15from . import case
16from .. import sql_db
18__unittest = True
20STDOUT_LINE = '\nStdout:\n%s'
21STDERR_LINE = '\nStderr:\n%s'
23ODOO_TEST_MAX_FAILED_TESTS = max(1, int(os.environ.get('ODOO_TEST_MAX_FAILED_TESTS', sys.maxsize)))
25stats_logger = logging.getLogger('odoo.tests.stats')
28class Stat(NamedTuple):
29 time: float = 0.0
30 queries: int = 0
32 def __add__(self, other: 'Stat') -> 'Stat':
33 if other == 0: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true
34 return self
36 if not isinstance(other, Stat): 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true
37 return NotImplemented
39 return Stat(
40 self.time + other.time,
41 self.queries + other.queries,
42 )
44_logger = logging.getLogger(__name__)
45_TEST_ID = re.compile(r"""
46^
47odoo\.addons\.
48(?P<module>[^.]+)
49\.tests\.
50(?P<class>.+)
51\.
52(?P<method>[^.]+)
53$
54""", re.VERBOSE)
57class OdooTestResult(object):
58 """
59 This class in inspired from TextTestResult and modifies TestResult
60 Instead of using a stream, we are using the logger.
62 unittest.TestResult: Holder for test result information.
64 Test results are automatically managed by the TestCase and TestSuite
65 classes, and do not need to be explicitly manipulated by writers of tests.
67 This version does not hold a list of failure but just a count since the failure is logged immediately
68 This version is also simplied to better match our use cases
69 """
71 _previousTestClass = None
72 _moduleSetUpFailed = False
74 def __init__(self, stream=None, descriptions=None, verbosity=None, global_report=None):
75 self.failures_count = 0
76 self.errors_count = 0
77 self.testsRun = 0
78 self.skipped = 0
79 self.tb_locals = False
80 # custom
81 self.time_start = None
82 self.queries_start = None
83 self._soft_fail = False
84 self.had_failure = False
85 self.stats = collections.defaultdict(Stat)
86 self.global_report = global_report
87 self.shouldStop = self.global_report and self.global_report.shouldStop or False
89 def total_errors_count(self):
90 result = self.errors_count + self.failures_count
91 if self.global_report:
92 result += self.global_report.total_errors_count()
93 return result
95 def _checkShouldStop(self):
96 if self.total_errors_count() >= ODOO_TEST_MAX_FAILED_TESTS:
97 global_report = self.global_report or self
98 if not global_report.shouldStop:
99 _logger.error(
100 "Test suite halted: max failed tests already reached (%s). "
101 "Remaining tests will be skipped.", ODOO_TEST_MAX_FAILED_TESTS)
102 global_report.shouldStop = True
103 self.shouldStop = True
105 def printErrors(self):
106 "Called by TestRunner after test run"
108 def startTest(self, test):
109 "Called when the given test is about to be run"
110 self.testsRun += 1
111 self.log(logging.INFO, 'Starting %s ...', self.getDescription(test), test=test)
112 self.time_start = time.time()
113 self.queries_start = sql_db.sql_counter
115 def stopTest(self, test):
116 """Called when the given test has been run"""
117 if stats_logger.isEnabledFor(logging.INFO): 117 ↛ exitline 117 didn't return from function 'stopTest' because the condition on line 117 was always true
118 self.stats[test.id()] = Stat(
119 time=time.time() - self.time_start,
120 queries=sql_db.sql_counter - self.queries_start,
121 )
123 def addError(self, test, err):
124 """Called when an error has occurred. 'err' is a tuple of values as
125 returned by sys.exc_info().
126 """
127 if self._soft_fail:
128 self.had_failure = True
129 else:
130 self.errors_count += 1
131 self.logError("ERROR", test, err)
132 self._checkShouldStop()
134 def addFailure(self, test, err):
135 """Called when an error has occurred. 'err' is a tuple of values as
136 returned by sys.exc_info()."""
137 if self._soft_fail:
138 self.had_failure = True
139 else:
140 self.failures_count += 1
141 self.logError("FAIL", test, err)
142 self._checkShouldStop()
144 def addSubTest(self, test, subtest, err):
145 if err is not None:
146 if issubclass(err[0], test.failureException):
147 self.addFailure(subtest, err)
148 else:
149 self.addError(subtest, err)
151 def addSuccess(self, test):
152 "Called when a test has completed successfully"
154 def addSkip(self, test, reason):
155 """Called when a test is skipped."""
156 self.skipped += 1
157 self.log(logging.INFO, 'skipped %s : %s', self.getDescription(test), reason, test=test)
159 def wasSuccessful(self):
160 """Tells whether or not this result was a success."""
161 # The hasattr check is for test_result's OldResult test. That
162 # way this method works on objects that lack the attribute.
163 # (where would such result intances come from? old stored pickles?)
164 return self.failures_count == self.errors_count == 0
166 def _exc_info_to_string(self, err, test):
167 """Converts a sys.exc_info()-style tuple of values into a string."""
168 exctype, value, tb = err
169 # Skip test runner traceback levels
170 while tb and self._is_relevant_tb_level(tb):
171 tb = tb.tb_next
173 if exctype is test.failureException:
174 # Skip assert*() traceback levels
175 length = self._count_relevant_tb_levels(tb)
176 else:
177 length = None
178 tb_e = traceback.TracebackException(
179 exctype, value, tb, limit=length, capture_locals=self.tb_locals)
180 msgLines = list(tb_e.format())
182 return ''.join(msgLines)
184 def _is_relevant_tb_level(self, tb):
185 return '__unittest' in tb.tb_frame.f_globals
187 def _count_relevant_tb_levels(self, tb):
188 length = 0
189 while tb and not self._is_relevant_tb_level(tb):
190 length += 1
191 tb = tb.tb_next
192 return length
194 def __repr__(self):
195 return f"<{self.__class__.__module__}.{self.__class__.__qualname__} run={self.testsRun} errors={self.errors_count} failures={self.failures_count}>"
197 def __str__(self):
198 return f'{self.failures_count} failed, {self.errors_count} error(s) of {self.testsRun} tests'
201 @contextlib.contextmanager
202 def soft_fail(self):
203 self.had_failure = False
204 self._soft_fail = True
205 try:
206 yield
207 finally:
208 self._soft_fail = False
209 self.had_failure = False
211 def update(self, other):
212 """ Merges an other test result into this one, only updates contents
214 :type other: OdooTestResult
215 """
216 self.failures_count += other.failures_count
217 self.errors_count += other.errors_count
218 self.testsRun += other.testsRun
219 self.skipped += other.skipped
220 self.stats.update(other.stats)
222 def log(self, level, msg, *args, test=None, exc_info=None, extra=None, stack_info=False, caller_infos=None):
223 """
224 ``test`` is the running test case, ``caller_infos`` is
225 (fn, lno, func, sinfo) (logger.findCaller format), see logger.log for
226 the other parameters.
227 """
228 test = test or self
229 while isinstance(test, case._SubTest) and test.test_case: 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true
230 test = test.test_case
231 logger = logging.getLogger(test.__module__)
232 try:
233 caller_infos = caller_infos or logger.findCaller(stack_info)
234 except ValueError:
235 caller_infos = "(unknown file)", 0, "(unknown function)", None
236 (fn, lno, func, sinfo) = caller_infos
237 # using logger.log makes it difficult to spot-replace findCaller in
238 # order to provide useful location information (the problematic spot
239 # inside the test function), so use lower-level functions instead
240 if logger.isEnabledFor(level): 240 ↛ exitline 240 didn't return from function 'log' because the condition on line 240 was always true
241 record = logger.makeRecord(logger.name, level, fn, lno, msg, args, exc_info, func, extra, sinfo)
242 logger.handle(record)
244 def log_stats(self):
245 if not stats_logger.isEnabledFor(logging.INFO): 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 return
248 details = stats_logger.isEnabledFor(logging.DEBUG)
249 stats_tree = collections.defaultdict(Stat)
250 counts = collections.Counter()
251 for test, stat in self.stats.items():
252 r = _TEST_ID.match(test)
253 if not r: # upgrade has tests at weird paths, ignore them 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 continue
256 stats_tree[r['module']] += stat
257 counts[r['module']] += 1
258 if details: 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 stats_tree['%(module)s.%(class)s' % r] += stat
260 stats_tree['%(module)s.%(class)s.%(method)s' % r] += stat
262 if details: 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true
263 stats_logger.debug('Detailed Tests Report:\n%s', ''.join(
264 f'\t{test}: {stats.time:.2f}s {stats.queries} queries\n'
265 for test, stats in sorted(stats_tree.items())
266 ))
267 else:
268 for module, stat in sorted(stats_tree.items()):
269 stats_logger.info(
270 "%s: %d tests %.2fs %d queries",
271 module, counts[module],
272 stat.time, stat.queries
273 )
275 def getDescription(self, test):
276 if isinstance(test, case._SubTest): 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 return 'Subtest %s.%s %s' % (test.test_case.__class__.__qualname__, test.test_case._testMethodName, test._subDescription())
278 if isinstance(test, case.TestCase): 278 ↛ 282line 278 didn't jump to line 282 because the condition on line 278 was always true
279 # since we have the module name in the logger, this will avoid to duplicate module info in log line
280 # we only apply this for TestCase since we can receive error handler or other special case
281 return "%s.%s" % (test.__class__.__qualname__, test._testMethodName)
282 return str(test)
284 @contextlib.contextmanager
285 def collectStats(self, test_id):
286 queries_before = sql_db.sql_counter
287 time_start = time.time()
289 yield
291 self.stats[test_id] += Stat(
292 time=time.time() - time_start,
293 queries=sql_db.sql_counter - queries_before,
294 )
296 def logError(self, flavour, test, error):
297 err = self._exc_info_to_string(error, test)
298 caller_infos = self.getErrorCallerInfo(error, test)
299 self.log(logging.INFO, '=' * 70, test=test, caller_infos=caller_infos) # keep this as info !!!!!!
300 self.log(logging.ERROR, "%s: %s\n%s", flavour, self.getDescription(test), err, test=test, caller_infos=caller_infos)
302 def getErrorCallerInfo(self, error, test):
303 """
304 :param error: A tuple (exctype, value, tb) as returned by sys.exc_info().
305 :param test: A TestCase that created this error.
306 :returns: a tuple (fn, lno, func, sinfo) matching the logger findCaller format or None
307 """
309 # only handle TestCase here. test can be an _ErrorHolder in some case (setup/teardown class errors)
310 if not isinstance(test, case.TestCase):
311 return
313 _, _, error_traceback = error
315 # move upwards the subtest hierarchy to find the real test
316 while isinstance(test, case._SubTest) and test.test_case:
317 test = test.test_case
319 method_tb = None
320 file_tb = None
321 filename = inspect.getfile(type(test))
323 # Note: since _ErrorCatcher was introduced, we could always take the
324 # last frame, keeping the check on the test method for safety.
325 # Fallbacking on file for cleanup file shoud always be correct to a
326 # minimal working version would be
327 #
328 # infos_tb = error_traceback
329 # while infos_tb.tb_next()
330 # infos_tb = infos_tb.tb_next()
331 #
332 while error_traceback:
333 code = error_traceback.tb_frame.f_code
334 if code.co_name in (test._testMethodName, 'setUp', 'tearDown'):
335 method_tb = error_traceback
336 if code.co_filename == filename:
337 file_tb = error_traceback
338 error_traceback = error_traceback.tb_next
340 infos_tb = method_tb or file_tb
341 if infos_tb:
342 code = infos_tb.tb_frame.f_code
343 lineno = infos_tb.tb_lineno
344 filename = code.co_filename
345 method = test._testMethodName
346 return (filename, lineno, method, None)