Belle II Software  release-05-01-25
test_backends.py
1 #!/usr/bin/env python3
2 # -*- coding: utf-8 -*-
3 
4 """
5 Script that tests the various `caf.backends` and some options that are available.
6 This cannot be run as a normal unit test because not all backends are available on the
7 DESY cloud servers. So the only think we can do is test them individually on machines that
8 do have the backends installed.
9 """
10 
11 import basf2
12 from basf2 import find_file
13 
14 # Prevent ROOT inserting its own help text and arguments
15 import ROOT
16 ROOT.PyConfig.IgnoreCommandLineOptions = True
17 
18 from pathlib import Path
19 
20 from caf.backends import Job, monitor_jobs, Local
21 from caf import cli
22 
23 test_script = Path(find_file("calibration/examples/job_submission/test_script.sh"))
24 test_basf2_script = Path(find_file("calibration/examples/job_submission/basic_basf2.py"))
25 test_data = Path(find_file("calibration/examples/job_submission/test_data"))
26 
27 
28 def test_path_exists(path):
29  if not path.exists():
30  raise FileNotFoundError(f"The expected file {test_script.as_posix()} does not exist.")
31 
32 
33 def create_jobs(args):
34  test_path_exists(test_data)
35  test_path_exists(test_script)
36  test_path_exists(test_basf2_script)
37  output_dir = Path("test_backends_jobs")
38  j1 = Job(name="test_job1")
39  # Working directory that will be sent to the backend and used as the current working directory
40  j1.working_dir = Path(output_dir, j1.name, "working_dir").absolute()
41  # Output directory, where the stdout and stderr will be sent.
42  j1.output_dir = Path(output_dir, j1.name, "output_dir").absolute()
43  # The command we want to run
44  j1.cmd = ["bash", test_script.name]
45  j1.args = ["first_arg_example", "\"Do quotes work?\""]
46  # We add the script we want to run to the input sandbox so that it is available locally in the working directory.
47  # We could instead simply change the 'cmd' above to always call the same file, but I prefer having the
48  # script copied into the working directory.
49  j1.input_sandbox_files.append(test_script.absolute())
50 
51  j2 = Job(name="test_job2")
52  # Working directory that will be sent to the backend and used as the current working directory
53  j2.working_dir = Path(output_dir, j2.name).absolute()
54  # Output directory, where the stdout and stderr will be sent. We have set this to be the same as the working dir
55  j2.output_dir = Path(output_dir, j2.name).absolute()
56  # The command we want to run
57  j2.cmd = ["basf2", test_basf2_script.name]
58  # Setup basf2 in the job the same way as our current basf2 environment
59  j2.append_current_basf2_setup_cmds()
60  j2.input_sandbox_files.append(test_basf2_script.absolute())
61  j2.input_files = sorted(p.as_posix() for p in Path(test_data).glob("*.root"))
62  if args.files_per_subjob:
63  j2.max_files_per_subjob = args.files_per_subjob
64  elif args.max_subjobs:
65  j2.max_subjobs = args.max_subjobs
66  return [j1, j2]
67 
68 
69 def get_argparser():
70  """Setup the argparser for this script"""
71  import argparse
72  parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
73 
74  subparsers = cli.add_backends_subparsers(parser)
75  for subparser in subparsers:
76  cli.add_basf2_options(subparser)
77  cli.add_monitor_options(subparser)
78  cli.add_job_options(subparser)
79  return parser
80 
81 
82 def main():
83  parser = get_argparser()
84  args = parser.parse_args()
85  basf2.set_log_level(basf2.LogLevel.names[args.log_level])
86  if args.debug_level:
87  basf2.set_log_level(basf2.LogLevel.DEBUG) # Override
88  basf2.set_debug_level(args.debug_level)
89  jobs = create_jobs(args)
90  backend = args.func(args)
91  backend.submit(jobs)
92  monitor_jobs(args, jobs)
93  if isinstance(backend, Local):
94  backend.join()
95 
96 
97 if __name__ == "__main__":
98  import sys
99  sys.exit(main())
cli.add_basf2_options
def add_basf2_options(parser, default_log_level="INFO")
Definition: cli.py:91
main
int main(int argc, char **argv)
Run all tests.
Definition: test_main.cc:77
cli.add_monitor_options
def add_monitor_options(parser, default_heartbeat=10)
Definition: cli.py:102
cli.add_backends_subparsers
def add_backends_subparsers(parser, default_max_processes=4, default_global_job_limit=Batch.default_global_job_limit, default_submission_check_heartbeat=Batch.default_sleep_between_submission_checks, local_func=command_local, lsf_func=command_lsf, pbs_func=command_pbs, condor_func=command_condor)
Definition: cli.py:127
cli.add_job_options
def add_job_options(parser)
Definition: cli.py:113