CMS 3D CMS Logo
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros 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 29 of file querying_tests.py.

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

Definition at line 56 of file querying_tests.py.

References data, reco.print(), and utils.to_datetime().

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

Variable Documentation

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

Definition at line 17 of file querying_tests.py.

querying_tests.secrets_source = None

Definition at line 18 of file querying_tests.py.

Generated for CMSSW Reference Manual by doxygen 1.8.5