12b2test_utils - Helper functions useful for test scripts
13-------------------------------------------------------
15This module contains functions which are commonly needed for tests like changing
16log levels or switching to an empty working directory
22from contextlib
import contextmanager
23from collections
import OrderedDict
28from b2test_utils
import logfilter
31def skip_test(reason, py_case=None):
32 """Skip a test script with a given reason. This function will end the script
35 This is intended for scripts to be run in :ref:`b2test-scripts
36 <b2test-scripts>` and will flag the script as skipped with the given reason
37 when tests are executed.
39 Useful if the test depends on some external condition like a web service and
40 missing this dependency should not fail the test run.
43 reason (str): the reason to skip the test.
44 py_case (unittest.TestCase): if this is to be skipped within python's
45 native unittest then pass the TestCase instance
48 py_case.skipTest(reason)
50 print(f
"TEST SKIPPED: {reason}", file=sys.stderr, flush=
True)
54def require_file(filename, data_type="", py_case=None):
55 """Check for the existence of a test input file before attempting to open it.
56 Skips the test if not found.
58 Wraps `basf2.find_file` for use in test scripts run as
59 :ref`b2test-scripts <b2test-scripts>`
62 filename (str): relative filename to look for, either in a central place or in the current working directory
63 data_type (str): case insensitive data type to find. Either empty string or one of ``"examples"`` or ``"validation"``.
64 py_case (unittest.TestCase): if this is to be skipped within python's native unittest then pass the TestCase instance
67 Full path to the test input file
70 fullpath = basf2.find_file(filename, data_type, silent=
False)
71 except FileNotFoundError
as fnf:
72 skip_test(f
'Cannot find: {fnf.filename}', py_case)
77def set_loglevel(loglevel):
79 temporarily set the log level to the specified `LogLevel <basf2.LogLevel>`. This returns a
80 context manager so it should be used in a ``with`` statement:
82 >>> with set_log_level(LogLevel.ERROR):
83 >>> # during this block the log level is set to ERROR
85 old_loglevel = basf2.logging.log_level
86 basf2.set_log_level(loglevel)
90 basf2.set_log_level(old_loglevel)
94def show_only_errors():
95 """temporarily set the log level to `ERROR <LogLevel.ERROR>`. This returns a
96 context manager so it should be used in a ``with`` statement
98 >>> with show_only_errors():
99 >>> B2INFO("this will not be shown")
100 >>> B2INFO("but this might")
102 with set_loglevel(basf2.LogLevel.ERROR):
106def configure_logging_for_tests(user_replacements=None, replace_cdb_provider=True):
108 Change the log system to behave a bit more appropriately for testing scenarios:
110 1. Simplify log message to be just ``[LEVEL] message``
111 2. Disable error summary, just additional noise
112 3. Intercept all log messages and replace
114 * the current working directory in log messaged with ``${cwd}``
115 * the current release version with ``${release_version}``
116 * the current default globaltags with ``${default_globaltag}``
117 * the contents of the following environment variables with their name
118 (or the listed replacement string):
120 - :envvar:`BELLE2_TOOLS`
121 - :envvar:`BELLE2_RELEASE_DIR` with ``BELLE2_SOFTWARE_DIR``
122 - :envvar:`BELLE2_LOCAL_DIR` with ``BELLE2_SOFTWARE_DIR``
123 - :envvar:`BELLE2_EXTERNALS_DIR`
124 - :envvar:`BELLE2_VALIDATION_DATA_DIR`
125 - :envvar:`BELLE2_EXAMPLES_DATA_DIR`
126 - :envvar:`BELLE2_BACKGROUND_DIR`
127 - :envvar:`BELLE2_CONDB_METADATA`
130 user_replacements (dict(str, str)): Additional strings and their replacements to replace in the output
131 replace_cdb_provider (bool): If False, it does not replace the conditions database metadata provider with
132 `BELLE2_CONDB_METADATA` (necessary for some specific tests)
135 This function should be called **after** switching directory to replace the correct directory name
137 .. versionadded:: release-04-00-00
139 basf2.logging.reset()
140 basf2.logging.enable_summary(
False)
141 basf2.logging.enable_python_logging =
True
142 basf2.logging.add_console()
145 for level
in basf2.LogLevel.values.values():
146 basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
152 replacements = OrderedDict()
157 replacements[
", ".join(basf2.conditions.default_globaltags)] =
"${default_globaltag}"
160 if replace_cdb_provider
and len(basf2.conditions.metadata_providers) > 0:
161 replacements[basf2.conditions.metadata_providers[0]] =
"${BELLE2_CONDB_METADATA}"
163 for env_name, replacement
in re.findall(
":envvar:`(.*?)`(?:.*``(.*?)``)?", configure_logging_for_tests.__doc__):
165 replacement = env_name
166 if env_name
in os.environ:
169 replacements[os.environ[env_name].rstrip(
'/ ')] = f
"${{{replacement}}}"
171 if user_replacements
is not None:
172 replacements.update(user_replacements)
174 replacements.setdefault(os.getcwd(),
"${cwd}")
175 sys.stdout = logfilter.LogReplacementFilter(sys.__stdout__, replacements)
179def working_directory(path):
180 """temporarily change the working directory to path
182 >>> with working_directory("testing"):
183 >>> # now in subdirectory "./testing/"
184 >>> # back to parent directory
186 This function will not create the directory for you. If changing into the
187 directory fails a `FileNotFoundError` will be raised.
189 dirname = os.getcwd()
198def clean_working_directory():
199 """Context manager to create a temporary directory and directly us it as
200 current working directory. The directory will automatically be deleted after
201 the with context is left.
203 >>> with clean_working_directory() as dirname:
204 >>> # now we are in an empty directory, name is stored in dirname
205 >>> assert(os.listdir() == [])
206 >>> # now we are back where we were before
208 with tempfile.TemporaryDirectory()
as tempdir:
209 with working_directory(tempdir):
214def local_software_directory():
215 """Context manager to make sure we are executed in the top software
216 directory by switching to $BELLE2_LOCAL_DIR or $BELLE2_RELEASE_DIR.
218 >>> with local_software_directory():
219 >>> assert(os.listdir().contains("analysis"))
221 directory = os.environ.get(
"BELLE2_LOCAL_DIR", os.environ.get(
"BELLE2_RELEASE_DIR",
None))
222 if directory
is None:
223 raise RuntimeError(
"Cannot find Belle II software directory, "
224 "have you setup the software correctly?")
226 with working_directory(directory):
230def run_in_subprocess(*args, target, **kwargs):
231 """Run the given ``target`` function in a child process using `multiprocessing.Process`
233 This avoids side effects: anything done in the target function will not
234 affect the current process. This is mostly useful for test scripts as
235 ``target`` can emit a `FATAL <LogLevel.FATAL>` error without killing script execution.
237 It will return the exitcode of the child process which should be 0 in case of no error
239 process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
242 return process.exitcode
245def safe_process(*args, **kwargs):
246 """Run `basf2.process` with the given path in a child process using
247 `multiprocessing.Process`
249 This avoids side effects (`safe_process` can be safely called multiple times)
250 and doesn't kill this script even if a segmentation violation or a `FATAL
251 <LogLevel.FATAL>` error occurs during processing.
253 It will return the exitcode of the child process which should be 0 in case of no error
255 return run_in_subprocess(target=basf2.process, *args, **kwargs)
258def check_error_free(tool, toolname, package, filter=lambda x:
False, toolopts=
None):
259 """Calls the ``tool`` with argument ``package`` and check that the output is
260 error-free. Optionally ``filter`` the output in case of error messages that
263 In case there is some output left, then prints the error message and exits
267 If the test is skipped or the test contains errors this function does
268 not return but will directly end the program.
271 tool(str): executable to call
272 toolname(str): human readable name of the tool
273 package(str): package to run over. Also the first argument to the tool
274 filter: function which gets called for each line of output and
275 if it returns True the line will be ignored.
276 toolopts(list(str)): extra options to pass to the tool.
279 if "BELLE2_LOCAL_DIR" not in os.environ
and "BELLE2_RELEASE_DIR" not in os.environ:
280 skip_test(
"No release is setup")
285 if package
is not None:
288 with local_software_directory():
290 output = subprocess.check_output(args, encoding=
"utf8")
291 except subprocess.CalledProcessError
as error:
293 output = error.output
295 clean_log = [e
for e
in output.splitlines()
if e
and not filter(e)]
296 if len(clean_log) > 0:
297 subject = f
"{package} package" if package
is not None else "repository"
299The {subject} has some {toolname} issues, which is now not allowed.
304and fix any issues you have introduced. Here is what {toolname} found:\n""")
305 print(
"\n".join(clean_log))
309def get_streamer_checksums(objects):
311 Extract the version and streamer checksum of the C++ objects in the given list
312 by writing them all to a TMemFile and getting back the streamer info list
313 automatically created by ROOT afterwards.
314 Please note, that this list also includes the streamer infos of all
315 base objects of the objects you gave.
317 Returns a dictionary object name -> (version, checksum).
323 f = ROOT.TMemFile(
"test_mem_file",
"RECREATE")
331 streamer_checksums = dict()
332 for streamer_info
in f.GetStreamerInfoList():
333 if not isinstance(streamer_info, ROOT.TStreamerInfo):
335 streamer_checksums[streamer_info.GetName()] = (streamer_info.GetClassVersion(), streamer_info.GetCheckSum())
338 return streamer_checksums
341def get_object_with_name(object_name, root=None):
343 (Possibly) recursively get the object with the given name from the Belle2 namespace.
345 If the object name includes a ".", the first part will be turned into an object (probably a module)
346 and the function is continued with this object as the root and the rest of the name.
348 If not, the object is extracted via a getattr call.
351 from ROOT
import Belle2
354 if "." in object_name:
355 namespace, object_name = object_name.split(
".", 1)
357 return get_object_with_name(object_name, get_object_with_name(namespace, root=root))
359 return getattr(root, object_name)
362def skip_test_if_light(py_case=None):
364 Skips the test if we are running in a light build (maybe this test tests
365 some generation example or whatever)
368 py_case (unittest.TestCase): if this is to be skipped within python's
369 native unittest then pass the TestCase instance
373 except ModuleNotFoundError:
374 skip_test(reason=
"We're in a light build.", py_case=py_case)
377def skip_test_if_central(py_case=None):
379 Skips the test if we are using a central release (and have no local
383 py_case (unittest.TestCase): if this is to be skipped within python's
384 native unittest then pass the TestCase instance
386 if "BELLE2_RELEASE_DIR" in os.environ:
387 skip_test(reason=
"We're in a central release.", py_case=py_case)
390def print_belle2_environment():
392 Prints all the BELLE2 environment variables on the screen.
394 basf2.B2INFO(
'The BELLE2 environment variables are:')
395 for key, value
in sorted(dict(os.environ).items()):
396 if 'BELLE2' in key.upper():
397 print(f
' {key}={value}')
401def temporary_environment(**environ):
403 Context manager that temporarily sets environment variables
404 for the current process. Inspired by https://stackoverflow.com/a/34333710
406 >>> with temporary_environment(BELLE2_TEMP_DIR='/tmp/belle2'):
407 ... "BELLE2_TEMP_DIR" in os.environ
410 >>> "BELLE2_TEMP_DIR" in os.environ
414 **environ: Arbitrary keyword arguments specifying environment
415 variables and their values.
417 old_environ = dict(os.environ)
418 os.environ.update(environ)
423 os.environ.update(old_environ)
428 Returns true if we are running a test on our CI system (currently GitLab pipeline).
429 The 'BELLE2_IS_CI' environment variable is set on CI only when the unit
432 return os.environ.get(
"BELLE2_IS_CI",
"no").lower()
in [
440def is_cdb_down() -> bool:
442 Returns true if the Conditions Database (CDB) is currently unavailable or slow to respond.
443 The 'BELLE2_IS_CDB_DOWN' environment variable can be used to dynamically exclude some
444 tests that rely on the CDB in case of problems.
446 return os.environ.get(
"BELLE2_IS_CDB_DOWN",
"no").lower()
in [