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