Belle II Software prerelease-10-00-00a
__init__.py
1#!/usr/bin/env python3
2
3
10
11"""
12b2test_utils - Helper functions useful for test scripts
13-------------------------------------------------------
14
15This module contains functions which are commonly needed for tests like changing
16log levels or switching to an empty working directory
17"""
18
19import sys
20import os
21import tempfile
22from contextlib import contextmanager
23from collections import OrderedDict
24import multiprocessing
25import basf2
26import subprocess
27import re
28from b2test_utils import logfilter
29
30
31def skip_test(reason, py_case=None):
32 """Skip a test script with a given reason. This function will end the script
33 and not return.
34
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.
38
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.
41
42 Parameters:
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
46 """
47 if py_case:
48 py_case.skipTest(reason)
49 else:
50 print(f"TEST SKIPPED: {reason}", file=sys.stderr, flush=True)
51 sys.exit(1)
52
53
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.
57
58 Wraps `basf2.find_file` for use in test scripts run as
59 :ref`b2test-scripts <b2test-scripts>`
60
61 Parameters:
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
65
66 Returns:
67 Full path to the test input file
68 """
69 try:
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)
73 return fullpath
74
75
76@contextmanager
77def set_loglevel(loglevel):
78 """
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:
81
82 >>> with set_log_level(LogLevel.ERROR):
83 >>> # during this block the log level is set to ERROR
84 """
85 old_loglevel = basf2.logging.log_level
86 basf2.set_log_level(loglevel)
87 try:
88 yield
89 finally:
90 basf2.set_log_level(old_loglevel)
91
92
93@contextmanager
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
97
98 >>> with show_only_errors():
99 >>> B2INFO("this will not be shown")
100 >>> B2INFO("but this might")
101 """
102 with set_loglevel(basf2.LogLevel.ERROR):
103 yield
104
105
106def configure_logging_for_tests(user_replacements=None):
107 """
108 Change the log system to behave a bit more appropriately for testing scenarios:
109
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
113
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):
119
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`
128
129 Parameters:
130 user_replacements (dict(str, str)): Additional strings and their replacements to replace in the output
131
132 Warning:
133 This function should be called **after** switching directory to replace the correct directory name
134
135 .. versionadded:: release-04-00-00
136 """
137 basf2.logging.reset()
138 basf2.logging.enable_summary(False)
139 basf2.logging.enable_python_logging = True
140 basf2.logging.add_console()
141 # clang prints namespaces differently so no function names. Also let's skip the line number,
142 # we don't want failing tests just because we added a new line of code. In fact, let's just see the message
143 for level in basf2.LogLevel.values.values():
144 basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
145
146 # now create dictionary of string replacements. Since each key can only be
147 # present once order is kind of important so the less portable ones like
148 # current directory should go first and might be overridden if for example
149 # the BELLE2_LOCAL_DIR is identical to the current working directory
150 replacements = OrderedDict()
151 try:
152 replacements[basf2.version.get_version()] = "${release_version}"
153 except Exception:
154 pass
155 replacements[", ".join(basf2.conditions.default_globaltags)] = "${default_globaltag}"
156 # add a special replacement for the CDB metadata provider URL, since it's not set via env. variable
157 replacements[basf2.conditions.default_metadata_provider_url] = "${BELLE2_CONDB_METADATA}"
158 # Let's be lazy and take the environment variables from the docstring so we don't have to repeat them here
159 for env_name, replacement in re.findall(":envvar:`(.*?)`(?:.*``(.*?)``)?", configure_logging_for_tests.__doc__):
160 if not replacement:
161 replacement = env_name
162 if env_name in os.environ:
163 # replace path from the environment with the name of the variable. But remove a trailing slash or whitespace so that
164 # the output doesn't depend on whether there is a tailing slash in the environment variable
165 replacements[os.environ[env_name].rstrip('/ ')] = f"${{{replacement}}}"
166
167 if user_replacements is not None:
168 replacements.update(user_replacements)
169 # add cwd only if it doesn't overwrite anything ...
170 replacements.setdefault(os.getcwd(), "${cwd}")
171 sys.stdout = logfilter.LogReplacementFilter(sys.__stdout__, replacements)
172
173
174@contextmanager
175def working_directory(path):
176 """temporarily change the working directory to path
177
178 >>> with working_directory("testing"):
179 >>> # now in subdirectory "./testing/"
180 >>> # back to parent directory
181
182 This function will not create the directory for you. If changing into the
183 directory fails a `FileNotFoundError` will be raised.
184 """
185 dirname = os.getcwd()
186 try:
187 os.chdir(path)
188 yield
189 finally:
190 os.chdir(dirname)
191
192
193@contextmanager
194def clean_working_directory():
195 """Context manager to create a temporary directory and directly us it as
196 current working directory. The directory will automatically be deleted after
197 the with context is left.
198
199 >>> with clean_working_directory() as dirname:
200 >>> # now we are in an empty directory, name is stored in dirname
201 >>> assert(os.listdir() == [])
202 >>> # now we are back where we were before
203 """
204 with tempfile.TemporaryDirectory() as tempdir:
205 with working_directory(tempdir):
206 yield tempdir
207
208
209@contextmanager
210def local_software_directory():
211 """Context manager to make sure we are executed in the top software
212 directory by switching to $BELLE2_LOCAL_DIR or $BELLE2_RELEASE_DIR.
213
214 >>> with local_software_directory():
215 >>> assert(os.listdir().contains("analysis"))
216 """
217 directory = os.environ.get("BELLE2_LOCAL_DIR", os.environ.get("BELLE2_RELEASE_DIR", None))
218 if directory is None:
219 raise RuntimeError("Cannot find Belle II software directory, "
220 "have you setup the software correctly?")
221
222 with working_directory(directory):
223 yield directory
224
225
226def run_in_subprocess(*args, target, **kwargs):
227 """Run the given ``target`` function in a child process using `multiprocessing.Process`
228
229 This avoids side effects: anything done in the target function will not
230 affect the current process. This is mostly useful for test scripts as
231 ``target`` can emit a `FATAL <LogLevel.FATAL>` error without killing script execution.
232
233 It will return the exitcode of the child process which should be 0 in case of no error
234 """
235 process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
236 process.start()
237 process.join()
238 return process.exitcode
239
240
241def safe_process(*args, **kwargs):
242 """Run `basf2.process` with the given path in a child process using
243 `multiprocessing.Process`
244
245 This avoids side effects (`safe_process` can be safely called multiple times)
246 and doesn't kill this script even if a segmentation violation or a `FATAL
247 <LogLevel.FATAL>` error occurs during processing.
248
249 It will return the exitcode of the child process which should be 0 in case of no error
250 """
251 return run_in_subprocess(target=basf2.process, *args, **kwargs)
252
253
254def check_error_free(tool, toolname, package, filter=lambda x: False, toolopts=None):
255 """Calls the ``tool`` with argument ``package`` and check that the output is
256 error-free. Optionally ``filter`` the output in case of error messages that
257 can be ignored.
258
259 In case there is some output left, then prints the error message and exits
260 (failing the test).
261
262 Warnings:
263 If the test is skipped or the test contains errors this function does
264 not return but will directly end the program.
265
266 Arguments:
267 tool(str): executable to call
268 toolname(str): human readable name of the tool
269 package(str): package to run over. Also the first argument to the tool
270 filter: function which gets called for each line of output and
271 if it returns True the line will be ignored.
272 toolopts(list(str)): extra options to pass to the tool.
273 """
274
275 if "BELLE2_LOCAL_DIR" not in os.environ and "BELLE2_RELEASE_DIR" not in os.environ:
276 skip_test("No release is setup")
277
278 args = [tool]
279 if toolopts:
280 args += toolopts
281 if package is not None:
282 args += [package]
283
284 with local_software_directory():
285 try:
286 output = subprocess.check_output(args, encoding="utf8")
287 except subprocess.CalledProcessError as error:
288 print(error)
289 output = error.output
290
291 clean_log = [e for e in output.splitlines() if e and not filter(e)]
292 if len(clean_log) > 0:
293 subject = f"{package} package" if package is not None else "repository"
294 print(f"""\
295The {subject} has some {toolname} issues, which is now not allowed.
296Please run:
297
298 $ {" ".join(args)}
299
300and fix any issues you have introduced. Here is what {toolname} found:\n""")
301 print("\n".join(clean_log))
302 sys.exit(1)
303
304
305def get_streamer_checksums(objects):
306 """
307 Extract the version and streamer checksum of the C++ objects in the given list
308 by writing them all to a TMemFile and getting back the streamer info list
309 automatically created by ROOT afterwards.
310 Please note, that this list also includes the streamer infos of all
311 base objects of the objects you gave.
312
313 Returns a dictionary object name -> (version, checksum).
314 """
315 # Always avoid the top-level 'import ROOT'.
316 import ROOT # noqa
317
318 # Write out the objects to a mem file
319 f = ROOT.TMemFile("test_mem_file", "RECREATE")
320 f.cd()
321
322 for o in objects:
323 o.Write()
324 f.Write()
325
326 # Go through all streamer infos and extract checksum and version
327 streamer_checksums = dict()
328 for streamer_info in f.GetStreamerInfoList():
329 if not isinstance(streamer_info, ROOT.TStreamerInfo):
330 continue
331 streamer_checksums[streamer_info.GetName()] = (streamer_info.GetClassVersion(), streamer_info.GetCheckSum())
332
333 f.Close()
334 return streamer_checksums
335
336
337def get_object_with_name(object_name, root=None):
338 """
339 (Possibly) recursively get the object with the given name from the Belle2 namespace.
340
341 If the object name includes a ".", the first part will be turned into an object (probably a module)
342 and the function is continued with this object as the root and the rest of the name.
343
344 If not, the object is extracted via a getattr call.
345 """
346 if root is None:
347 from ROOT import Belle2 # noqa
348 root = Belle2
349
350 if "." in object_name:
351 namespace, object_name = object_name.split(".", 1)
352
353 return get_object_with_name(object_name, get_object_with_name(namespace, root=root))
354
355 return getattr(root, object_name)
356
357
358def skip_test_if_light(py_case=None):
359 """
360 Skips the test if we are running in a light build (maybe this test tests
361 some generation example or whatever)
362
363 Parameters:
364 py_case (unittest.TestCase): if this is to be skipped within python's
365 native unittest then pass the TestCase instance
366 """
367 try:
368 import generators # noqa
369 except ModuleNotFoundError:
370 skip_test(reason="We're in a light build.", py_case=py_case)
371
372
373def skip_test_if_central(py_case=None):
374 """
375 Skips the test if we are using a central release (and have no local
376 git repository)
377
378 Parameters:
379 py_case (unittest.TestCase): if this is to be skipped within python's
380 native unittest then pass the TestCase instance
381 """
382 if "BELLE2_RELEASE_DIR" in os.environ:
383 skip_test(reason="We're in a central release.", py_case=py_case)
384
385
386def print_belle2_environment():
387 """
388 Prints all the BELLE2 environment variables on the screen.
389 """
390 basf2.B2INFO('The BELLE2 environment variables are:')
391 for key, value in sorted(dict(os.environ).items()):
392 if 'BELLE2' in key.upper():
393 print(f' {key}={value}')
394
395
396@contextmanager
397def temporary_environment(**environ):
398 """
399 Context manager that temporarily sets environment variables
400 for the current process. Inspired by https://stackoverflow.com/a/34333710
401
402 >>> with temporary_environment(BELLE2_TEMP_DIR='/tmp/belle2'):
403 ... "BELLE2_TEMP_DIR" in os.environ
404 True
405
406 >>> "BELLE2_TEMP_DIR" in os.environ
407 False
408
409 Args:
410 **environ: Arbitrary keyword arguments specifying environment
411 variables and their values.
412 """
413 old_environ = dict(os.environ)
414 os.environ.update(environ)
415 try:
416 yield
417 finally:
418 os.environ.clear()
419 os.environ.update(old_environ)
420
421
422def is_ci() -> bool:
423 """
424 Returns true if we are running a test on our CI system (currently GitLab pipeline).
425 The 'BELLE2_IS_CI' environment variable is set on CI only when the unit
426 tests are run.
427 """
428 return os.environ.get("BELLE2_IS_CI", "no").lower() in [
429 "yes",
430 "1",
431 "y",
432 "on",
433 ]
434
435
436def is_cdb_down() -> bool:
437 """
438 Returns true if the Conditions Database (CDB) is currently unavailable or slow to respond.
439 The 'BELLE2_IS_CDB_DOWN' environment variable can be used to dynamically exclude some
440 tests that rely on the CDB in case of problems.
441 """
442 return os.environ.get("BELLE2_IS_CDB_DOWN", "no").lower() in [
443 "yes",
444 "1",
445 "y",
446 "on",
447 ]