CMS 3D CMS Logo

 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Groups Pages
Classes | Functions | Variables
querying_tests Namespace Reference

Classes

class  global_tag_map_tests
 
class  iov_tests
 
class  misc_tests
 
class  payload_tests
 
class  querying_tests
 
class  result_type_tests
 
class  tag_tests
 

Functions

def factory_tests
 
def global_tag_tests
 

Variables

string prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"
 
 secrets_source = None
 

Function Documentation

def querying_tests.factory_tests (   querying_tests)
Factory

Definition at line 30 of file querying_tests.py.

30 
31 def factory_tests(querying_tests):
32 
33  """
34  Factory
35  """
36 
37  def test_check_keys_in_models(self):
38  """
39  Verifies that each key that is required is present in self.connection's model dictionary,
40  and also that self.connection has a proxy method for each model that is generated.
41  """
42  keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
43  for key in keys:
44  self.assertTrue(key in self.connection.models.keys())
45  proxy_method_names = ["global_tag", "global_tag_map", "tag", "iov", "payload"]
46  for name in proxy_method_names:
47  self.assertTrue(hasattr(self.connection, name))
48 
49  def test_raw_factory_calls_all_models(self):
50  """
51  Verifies that the factory object held by the connection generates classes belonging to the type
52  held by self.connection.
53  """
54  keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
55  for key in keys:
56  self.assertTrue(isinstance(self.connection.factory.object(key), self.connection.models[key]))
def querying_tests.global_tag_tests (   querying_tests)
Global Tags

Definition at line 57 of file querying_tests.py.

References data, print(), sistrip::SpyUtilities.range(), and utils.to_datetime().

57 
58 def global_tag_tests(querying_tests):
59 
60  """
61  Global Tags
62  """
63 
64  def test_get_global_tag(self):
65  # hard coded global tag for now
66  global_tag = self.connection.global_tag(name=self.global_tag_name)
67  self.assertTrue(global_tag.name != None)
68  self.assertTrue(global_tag.tags() != None)
69  self.assertTrue(isinstance(global_tag, self.connection.models["globaltag"]))
70 
71  def test_get_empty_global_tag(self):
72  empty_gt = self.connection.global_tag()
73  self.assertTrue(isinstance(empty_gt, self.connection.models["globaltag"]))
74  self.assertTrue(empty_gt.empty)
75 
76  def test_all_global_tags_empty_gt(self):
77  empty_gt = self.connection.global_tag()
78  all_gts = empty_gt.all(amount=10)
79  self.assertTrue(all_gts != None)
80  self.assertTrue(len(all_gts.data()) != 0)
81 
82  def test_all_method_parameters(self):
83  if self.connection.connection_data["host"].lower() == "frontier":
84  print("Cannot query for timestamps on frontier connection.")
85  return
86  empty_gt = self.connection.global_tag()
87  sample_time = datetime.datetime(year=2016, month=1, day=1)
88  now = datetime.datetime.now()
89  time_range = Range(sample_time, now)
90  time_radius = Radius(sample_time, datetime.timedelta(weeks=4))
91  all_gts_in_interval = empty_gt.all(insertion_time=time_range).data()
92  for gt in all_gts_in_interval:
93  self.assertTrue(sample_time <= gt.insertion_time <= now)
94 
95  def test_as_dicts_method_without_timestamps_conversion(self):
96  global_tag = self.connection.global_tag(name=self.global_tag_name)
97  dict_form = global_tag.as_dicts(convert_timestamps=False)
98  self.assertTrue(isinstance(dict_form, dict))
99  self.assertTrue(dict_form["insertion_time"], datetime.datetime)
100  self.assertTrue(dict_form["snapshot_time"], datetime.datetime)
101 
102  def test_as_dicts_method_with_timestamps_conversion(self):
103  global_tag = self.connection.global_tag(name=self.global_tag_name)
104  dict_form = global_tag.as_dicts(convert_timestamps=True)
105  self.assertTrue(isinstance(dict_form, dict))
106  self.assertTrue(dict_form["insertion_time"], str)
107  self.assertTrue(dict_form["snapshot_time"], str)
108 
109  # now check that the timestamp encoded in the strings is the same as the
110  # datetime object that were converted
111  self.assertTrue(to_datetime(dict_form["insertion_time"]) == global_tag.insertion_time)
112  self.assertTrue(to_datetime(dict_form["snapshot_time"]) == global_tag.snapshot_time)
113 
114  def test_get_tag_maps_in_global_tag(self):
115  global_tag = self.connection.global_tag(name=self.global_tag_name)
116 
117  # get global tag maps
118  global_tag_maps = global_tag.tags()
119  self.assertTrue(isinstance(global_tag_maps, data_sources.json_list))
120  global_tag_maps_list = global_tag_maps.data()
121  self.assertTrue(isinstance(global_tag_maps_list, list))
122  self.assertTrue(len(global_tag_maps_list) != 0)
123 
124  for gt_map in global_tag_maps_list:
125  self.assertTrue(isinstance(gt_map, self.connection.models["globaltagmap"]))
126 
127  def test_get_tag_maps_in_global_tag_with_parameters(self):
128  global_tag = self.connection.global_tag(name=self.global_tag_name)
129 
130  global_tag_maps_specific_record = global_tag.tags(record="AlCaRecoTriggerBitsRcd")
131 
132  self.assertTrue(isinstance(global_tag_maps_specific_record, data_sources.json_list))
133 
134  gt_maps_spec_record_list = global_tag_maps_specific_record.data()
135  self.assertTrue(isinstance(gt_maps_spec_record_list, list))
136  # this global tag is old, so this test is unlikely to fail since
137  # its global tag maps will not fail
138  self.assertTrue(len(gt_maps_spec_record_list) == 3)
139 
140  def test_get_all_iovs_within_range(self):
141  global_tag = self.connection.global_tag(name=self.global_tag_name)
142 
143  since_range = self.connection.range(200000, 300000)
144  iovs = global_tag.iovs(since=since_range)
145  self.assertTrue(isinstance(iovs, data_sources.json_list))
146  iovs_list = iovs.data()
147  self.assertTrue(isinstance(iovs_list, list))
148 
149  for iov in iovs_list:
150  self.assertTrue(isinstance(iov, self.connection.models["iov"]))
151  self.assertTrue(200000 <= iov.since <= 300000)
152 
153  def test_gt_diff(self):
154  gt1 = self.connection.global_tag(name="74X_dataRun1_v1")
155  gt2 = self.connection.global_tag(name="74X_dataRun1_v3")
156  difference_by_arithmetic = gt1 - gt2
157  difference_by_method = gt1.diff(gt2)
158 
159  # verify both methods of difference
160  self.assertTrue(isinstance(difference_by_arithmetic, data_sources.json_list))
161  self.assertTrue(isinstance(difference_by_method, data_sources.json_list))
162 
163  difference_arithmetic_list = difference_by_arithmetic.data()
164  difference_method_list = difference_by_method.data()
165  self.assertTrue(isinstance(difference_arithmetic_list, list))
166  self.assertTrue(isinstance(difference_method_list, list))
167 
168  self.assertTrue(len(difference_arithmetic_list) == len(difference_method_list))
169 
170  for n in range(len(difference_arithmetic_list)):
171  self.assertTrue(difference_arithmetic_list[n]["%s Tag" % gt1.name] != difference_arithmetic_list[n]["%s Tag" % gt2.name])
172  self.assertTrue(difference_method_list[n]["%s Tag" % gt1.name] != difference_method_list[n]["%s Tag" % gt2.name])
def to_datetime
Definition: utils.py:12
PixelRecoRange< float > Range
const uint16_t range(const Frame &aFrame)
void print(TMatrixD &m, const char *label=nullptr, bool mathematicaFormat=false)
Definition: Utilities.cc:47
char data[epos_bytes_allocation]
Definition: EPOS_Wrapper.h:79

Variable Documentation

string querying_tests.prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"

Definition at line 18 of file querying_tests.py.

querying_tests.secrets_source = None

Definition at line 19 of file querying_tests.py.