Belle II Software light-2406-ragdoll
check_payload_metadata.py
1
8'''
9Check the revision number of a payload used by a module.
10'''
11
12import os
13
14import basf2 as b2
15import b2test_utils as b2tu
16import ROOT.Belle2 as B2
17
18
19class CheckPayloadMetadata(b2.Module):
20 '''
21 Small module for checking some payload metadata.
22 '''
23
24 def __init__(self, use_testing_payloads, md5_revision):
25 '''Consructor.'''
26 super().__init__()
27
28 self.beam_parameters = B2.PyDBObj('BeamParameters')
29
30 self.use_testing_payloads = use_testing_payloads
31
32 self.md5_revision = md5_revision
33
34 def beginRun(self):
35 '''Begin run.'''
36 assert(self.beam_parameters.isValid())
38 assert('testingdb/testingdb.txt' in self.beam_parameters.getGlobaltag())
39 assert(self.beam_parameters.getRevision() == 0)
40 assert(os.path.basename(self.beam_parameters.getFilename()).startswith('dbstore'))
41 assert(os.path.basename(self.beam_parameters.getFilename()).index(self.md5_revision) != -1)
42 assert(self.beam_parameters.getIoV() == B2.IntervalOfValidity(0, 1, 2, 3))
43 else:
44 assert(self.beam_parameters.getGlobaltag() == b2.conditions.default_globaltags[0])
45 assert(self.beam_parameters.getRevision() > 0)
46 assert(self.beam_parameters.getFilename().startswith('/cvmfs'))
47 assert(self.beam_parameters.getIoV() == B2.IntervalOfValidity(0, 0, 0, -1))
48
49
50def create_beam_parameters_payload(testing_database_path):
51 '''
52 Create locally a BeamParameters payload.
53 '''
54 b2.conditions.expert_settings(save_payloads=testing_database_path)
55 beam_parameters = B2.BeamParameters()
56 database = B2.Database.Instance()
57 iov = B2.IntervalOfValidity(0, 1, 2, 3)
58 assert(database.storeData('BeamParameters', beam_parameters, iov))
59
60
61def testing_path(use_testing_payloads=False, remove_default_globaltag=False, testing_database_path=None, md5_revision=None):
62 '''
63 The basf2 steering path for this test.
64 '''
65 if remove_default_globaltag:
66 b2.conditions.override_globaltags()
67 if use_testing_payloads:
68 b2.conditions.prepend_testing_payloads(b2.find_file(testing_database_path))
69 testing_path = b2.Path()
70 testing_path.add_module('EventInfoSetter', expList=[0], runList=[1])
71 testing_path.add_module(CheckPayloadMetadata(use_testing_payloads=use_testing_payloads, md5_revision=md5_revision))
72 b2.process(testing_path)
73
74
75if __name__ == '__main__':
76
77 with b2tu.clean_working_directory():
78
79
80 testing_database_path = 'testingdb/testingdb.txt'
81
82 # Create the testing payloads.
83 assert(b2tu.run_in_subprocess(target=create_beam_parameters_payload, testing_database_path=testing_database_path) == 0)
84
85 # Identify the md5 revision string of the payload.
86
87 md5_revision = None
88 with open(b2.find_file('testingdb/testingdb.txt')) as testing_database_file:
89 for line in testing_database_file:
90 md5_revision = str(line.split(' ')[1])
91
92 # Check if the process aborts because no valid payloads are found.
93 assert(b2tu.run_in_subprocess(target=testing_path,
94 use_testing_payloads=False,
95 remove_default_globaltag=True,
96 ) == 1)
97
98 # Check if the default globaltag is correctly used.
99 assert(b2tu.run_in_subprocess(target=testing_path,
100 use_testing_payloads=False,
101 remove_default_globaltag=False,
102 ) == 0)
103
104 # Check if the testing payload is correctly used.
105 assert(b2tu.run_in_subprocess(target=testing_path,
106 use_testing_payloads=True,
107 remove_default_globaltag=True,
108 testing_database_path=testing_database_path,
109 md5_revision=md5_revision,
110 ) == 0)
111
112 # Check if the testing payload is used instead of the default globaltag.
113 assert(b2tu.run_in_subprocess(target=testing_path,
114 use_testing_payloads=True,
115 remove_default_globaltag=False,
116 testing_database_path=testing_database_path,
117 md5_revision=md5_revision,
118 ) == 0)
md5_revision
md5 revision hash of the payload.
def __init__(self, use_testing_payloads, md5_revision)
beam_parameters
BeamParameters database object.
use_testing_payloads
Flag to mark if testing payloads are used or not.