Belle II Software light-2406-ragdoll
steering_files.py
1#!/usr/bin/env python3
2
3
10
11"""
12Test all the steering files used in the online_book lessons.
13Proudly based on analysis/test/examples.py.
14"""
15
16# std
17import os
18import sys
19import subprocess
20import unittest
21import glob
22import shutil
23from typing import Optional, List, Dict
24from pathlib import Path
25
26# 3rd
27from ROOT import TFile
28
29# basf2
30from basf2 import find_file
31from b2test_utils import clean_working_directory, is_ci
32
33
34def light_release() -> bool:
35 """Returns true if we're in a light release"""
36 try:
37 # pylint: disable=import-outside-toplevel
38 # pylint: disable=unused-import
39 import generators # noqa
40 except ModuleNotFoundError:
41 return True
42 return False
43
44
45def _touch_file_default(path: str):
46 Path(path).touch()
47
48
49def _touch_file_with_root(path: str) -> None:
50 f = TFile(path, "NEW")
51 f.Close()
52 assert Path(path).is_file()
53
54
55def _touch_file_with_subprocess(path: str) -> None:
56 subprocess.run(["touch", path]).check_returncode()
57
58
59def _touch_file_with_subprocess_and_root(path: str) -> None:
60 filename = Path(path).name
61 working_dir = Path(path).parent
62 cmd = ["root", "-x", "-l", "-q", "-e", f"TFile f(\"{filename}\", \"NEW\"); if (not f.IsOpen()) gSystem->Exit(1);"]
63 subprocess.run(cmd, cwd=working_dir).check_returncode()
64
65
66def _touch_file_test(method, path: str, **kwargs):
67 try:
68 method(path, **kwargs)
69 except Exception as e:
70 print(f"{method.__name__}: Tried to touch file with, but failed: {e}")
71 else:
72 print(f"{method.__name__}: Successfully touched file")
73 Path(path).unlink()
74
75
76def _permission_report(folder: str) -> None:
77 """Quick helper function to show permissions of folder and a selection
78 of files in it
79 """
80 folder = Path(folder)
81 print("-" * 80)
82 print(f"Permissions of {folder}: {folder.stat()}")
83 content = list(folder.iterdir())
84 if content:
85 print(
86 f"Permission of one of its contents. {content[0]}: "
87 f"{content[0].stat()}"
88 )
89 test_file = folder / "Bd2JpsiKS.root"
90 methods = [
91 _touch_file_default,
92 _touch_file_with_root,
93 _touch_file_with_subprocess,
94 _touch_file_with_subprocess_and_root
95 ]
96 for method in methods:
97 _touch_file_test(method, str(test_file))
98 print("-" * 80)
99
100
101class SteeringFileTest(unittest.TestCase):
102 """Test steering files"""
103
105 self,
106 path_to_glob: str,
107 broken: Optional[List[str]] = None,
108 additional_arguments: Optional[List[str]] = None,
109 expensive_tests: Optional[List[str]] = None,
110 skip_in_light: Optional[List[str]] = None,
111 skip: Optional[List[str]] = None,
112 n_events: Optional[Dict[str, int]] = None,
113 ):
114 """
115 Internal function to test a directory full of example scripts with an
116 optional list of broken scripts to be skipped.
117
118 Parameters:
119 path_to_glob (str): the path to a directory to search for python
120 scripts (must end in .py)
121 broken (list(str)): (optional) names of scripts that are known to
122 be broken and can be skipped
123 additional_arguments (list(str)): (optional) additional arguments
124 for basf2 to be passed when testing the scripts
125 expensive_tests (list(str)): (optional) names of scripts that take
126 longer and should e.g. not run in GitLab pipeline
127 skip_in_light (list(str)): (optional) names of scripts that have to
128 be excluded in light builds
129 skip (list(str)): (optional) names of scripts to always skip
130 n_events (dict(str, int)): mapping of name of script to number of
131 required events for it to run (`-n` argument). If a filename
132 isn't listed, we assume 1
133 """
134 if additional_arguments is None:
135 additional_arguments = []
136 if broken is None:
137 broken = []
138 if expensive_tests is None:
139 expensive_tests = []
140 if skip_in_light is None:
141 skip_in_light = []
142 if skip is None:
143 skip = []
144 if n_events is None:
145 n_events = {}
146 # we have to copy all the steering files (plus other stuffs, like decfiles) we want to test
147 # into a new directory and then cd it as working directory when subprocess.run is executed,
148 # otherwise the test will fail horribly if find_file is called by one of the tested steerings.
149 original_dir = find_file(path_to_glob)
150 print(f"Our user id: {os.getuid()}")
151 _permission_report(original_dir)
152 working_dir = find_file(shutil.copytree(original_dir, "working_dir"))
153 _permission_report(working_dir)
154 # Add write permissions for user to this directory
155 os.chmod(working_dir, 0o744)
156 _permission_report(working_dir)
157 all_egs = sorted(glob.glob(working_dir + "/*.py"))
158 for eg in all_egs:
159 filename = os.path.basename(eg)
160 if filename in broken:
161 continue
162 if is_ci() and filename in expensive_tests:
163 continue
164 if light_release() and filename in skip_in_light:
165 continue
166 if filename in skip:
167 continue
168 with self.subTest(msg=filename):
169 # pylint: disable=subprocess-run-check
170 result = subprocess.run(
171 [
172 "basf2",
173 "-n",
174 str(n_events.get(filename, 1)),
175 eg,
176 *additional_arguments,
177 ],
178 stdout=subprocess.PIPE,
179 stderr=subprocess.STDOUT,
180 cwd=working_dir,
181 )
182 if result.returncode != 0:
183 # failure running example so let's print the output
184 # on stderr so it's not split from output of unittest
185 # done like this since we don't want to decode/encode utf8
186 sys.stdout.buffer.write(result.stdout)
187 self.assertEqual(result.returncode, 0)
188
189 # fixme: This should be made to run on buildbot, i.e. by adding the/some
190 # files to the examples/validation directory
191 @unittest.skipIf(
192 not os.path.exists(
193 find_file(
194 "starterkit/2021/1111540100_eph3_BGx0_1.root",
195 "examples",
196 silent=True,
197 )
198 ),
199 "Test data files not found.",
200 )
202 """Test lesson on basf2 basics."""
204 path_to_glob="online_book/basf2/steering_files",
205 additional_arguments=["1"],
206 expensive_tests=["065_generate_mc.py", "067_generate_mc.py"],
207 skip_in_light=[
208 "065_generate_mc.py",
209 "067_generate_mc.py",
210 "085_module.py",
211 "087_module.py",
212 ],
213 n_events={
214 # See https://questions.belle2.org/question/11344/
215 "091_cs.py": 3000,
216 },
217 )
218
219
220if __name__ == "__main__":
221 with clean_working_directory():
222 unittest.main()
def _test_examples_dir(self, str path_to_glob, Optional[List[str]] broken=None, Optional[List[str]] additional_arguments=None, Optional[List[str]] expensive_tests=None, Optional[List[str]] skip_in_light=None, Optional[List[str]] skip=None, Optional[Dict[str, int]] n_events=None)