File indexing completed on 2024-11-25 02:29:20
0001 import unittest
0002 import sys
0003 import datetime
0004 import pprint
0005 import subprocess
0006 import os
0007
0008 import CondCore.Utilities.CondDBFW.querying as querying
0009 import CondCore.Utilities.CondDBFW.data_sources as data_sources
0010 import CondCore.Utilities.CondDBFW.data_formats as data_formats
0011 import CondCore.Utilities.CondDBFW.shell as shell
0012 import CondCore.Utilities.CondDBFW.models as models
0013
0014 from CondCore.Utilities.CondDBFW.utils import to_timestamp, to_datetime, friendly_since
0015 from CondCore.Utilities.CondDBFW.models import Range, Radius
0016
0017 prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"
0018 secrets_source = None
0019
0020 class querying_tests(unittest.TestCase):
0021 def setUp(self):
0022 self.connection = querying.connect(prod_connection_string, secrets=secrets_source)
0023 self.global_tag_name = "74X_dataRun1_HLT_frozen_v2"
0024 def test_check_connection(self):
0025 self.assertTrue(self.connection != None)
0026 def tearDown(self):
0027 self.connection.tear_down()
0028
0029 def factory_tests(querying_tests):
0030
0031 """
0032 Factory
0033 """
0034
0035 def test_check_keys_in_models(self):
0036 """
0037 Verifies that each key that is required is present in self.connection's model dictionary,
0038 and also that self.connection has a proxy method for each model that is generated.
0039 """
0040 keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
0041 for key in keys:
0042 self.assertTrue(key in self.connection.models.keys())
0043 proxy_method_names = ["global_tag", "global_tag_map", "tag", "iov", "payload"]
0044 for name in proxy_method_names:
0045 self.assertTrue(hasattr(self.connection, name))
0046
0047 def test_raw_factory_calls_all_models(self):
0048 """
0049 Verifies that the factory object held by the connection generates classes belonging to the type
0050 held by self.connection.
0051 """
0052 keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
0053 for key in keys:
0054 self.assertTrue(isinstance(self.connection.factory.object(key), self.connection.models[key]))
0055
0056 def global_tag_tests(querying_tests):
0057
0058 """
0059 Global Tags
0060 """
0061
0062 def test_get_global_tag(self):
0063
0064 global_tag = self.connection.global_tag(name=self.global_tag_name)
0065 self.assertTrue(global_tag.name != None)
0066 self.assertTrue(global_tag.tags() != None)
0067 self.assertTrue(isinstance(global_tag, self.connection.models["globaltag"]))
0068
0069 def test_get_empty_global_tag(self):
0070 empty_gt = self.connection.global_tag()
0071 self.assertTrue(isinstance(empty_gt, self.connection.models["globaltag"]))
0072 self.assertTrue(empty_gt.empty)
0073
0074 def test_all_global_tags_empty_gt(self):
0075 empty_gt = self.connection.global_tag()
0076 all_gts = empty_gt.all(amount=10)
0077 self.assertTrue(all_gts != None)
0078 self.assertTrue(len(all_gts.data()) != 0)
0079
0080 def test_all_method_parameters(self):
0081 if self.connection.connection_data["host"].lower() == "frontier":
0082 print("Cannot query for timestamps on frontier connection.")
0083 return
0084 empty_gt = self.connection.global_tag()
0085 sample_time = datetime.datetime(year=2016, month=1, day=1)
0086 now = datetime.datetime.now()
0087 time_range = Range(sample_time, now)
0088 time_radius = Radius(sample_time, datetime.timedelta(weeks=4))
0089 all_gts_in_interval = empty_gt.all(insertion_time=time_range).data()
0090 for gt in all_gts_in_interval:
0091 self.assertTrue(sample_time <= gt.insertion_time <= now)
0092
0093 def test_as_dicts_method_without_timestamps_conversion(self):
0094 global_tag = self.connection.global_tag(name=self.global_tag_name)
0095 dict_form = global_tag.as_dicts(convert_timestamps=False)
0096 self.assertTrue(isinstance(dict_form, dict))
0097 self.assertTrue(dict_form["insertion_time"], datetime.datetime)
0098 self.assertTrue(dict_form["snapshot_time"], datetime.datetime)
0099
0100 def test_as_dicts_method_with_timestamps_conversion(self):
0101 global_tag = self.connection.global_tag(name=self.global_tag_name)
0102 dict_form = global_tag.as_dicts(convert_timestamps=True)
0103 self.assertTrue(isinstance(dict_form, dict))
0104 self.assertTrue(dict_form["insertion_time"], str)
0105 self.assertTrue(dict_form["snapshot_time"], str)
0106
0107
0108
0109 self.assertTrue(to_datetime(dict_form["insertion_time"]) == global_tag.insertion_time)
0110 self.assertTrue(to_datetime(dict_form["snapshot_time"]) == global_tag.snapshot_time)
0111
0112 def test_get_tag_maps_in_global_tag(self):
0113 global_tag = self.connection.global_tag(name=self.global_tag_name)
0114
0115
0116 global_tag_maps = global_tag.tags()
0117 self.assertTrue(isinstance(global_tag_maps, data_sources.json_list))
0118 global_tag_maps_list = global_tag_maps.data()
0119 self.assertTrue(isinstance(global_tag_maps_list, list))
0120 self.assertTrue(len(global_tag_maps_list) != 0)
0121
0122 for gt_map in global_tag_maps_list:
0123 self.assertTrue(isinstance(gt_map, self.connection.models["globaltagmap"]))
0124
0125 def test_get_tag_maps_in_global_tag_with_parameters(self):
0126 global_tag = self.connection.global_tag(name=self.global_tag_name)
0127
0128 global_tag_maps_specific_record = global_tag.tags(record="AlCaRecoTriggerBitsRcd")
0129
0130 self.assertTrue(isinstance(global_tag_maps_specific_record, data_sources.json_list))
0131
0132 gt_maps_spec_record_list = global_tag_maps_specific_record.data()
0133 self.assertTrue(isinstance(gt_maps_spec_record_list, list))
0134
0135
0136 self.assertTrue(len(gt_maps_spec_record_list) == 3)
0137
0138 def test_get_all_iovs_within_range(self):
0139 global_tag = self.connection.global_tag(name=self.global_tag_name)
0140
0141 since_range = self.connection.range(200000, 300000)
0142 iovs = global_tag.iovs(since=since_range)
0143 self.assertTrue(isinstance(iovs, data_sources.json_list))
0144 iovs_list = iovs.data()
0145 self.assertTrue(isinstance(iovs_list, list))
0146
0147 for iov in iovs_list:
0148 self.assertTrue(isinstance(iov, self.connection.models["iov"]))
0149 self.assertTrue(200000 <= iov.since <= 300000)
0150
0151 def test_gt_diff(self):
0152 gt1 = self.connection.global_tag(name="74X_dataRun1_v1")
0153 gt2 = self.connection.global_tag(name="74X_dataRun1_v3")
0154 difference_by_arithmetic = gt1 - gt2
0155 difference_by_method = gt1.diff(gt2)
0156
0157
0158 self.assertTrue(isinstance(difference_by_arithmetic, data_sources.json_list))
0159 self.assertTrue(isinstance(difference_by_method, data_sources.json_list))
0160
0161 difference_arithmetic_list = difference_by_arithmetic.data()
0162 difference_method_list = difference_by_method.data()
0163 self.assertTrue(isinstance(difference_arithmetic_list, list))
0164 self.assertTrue(isinstance(difference_method_list, list))
0165
0166 self.assertTrue(len(difference_arithmetic_list) == len(difference_method_list))
0167
0168 for n in range(len(difference_arithmetic_list)):
0169 self.assertTrue(difference_arithmetic_list[n]["%s Tag" % gt1.name] != difference_arithmetic_list[n]["%s Tag" % gt2.name])
0170 self.assertTrue(difference_method_list[n]["%s Tag" % gt1.name] != difference_method_list[n]["%s Tag" % gt2.name])
0171
0172 class global_tag_map_tests(querying_tests):
0173
0174
0175 """
0176 Global Tag Map
0177 """
0178
0179 def test_get_global_tag_map(self):
0180 global_tag_name = "74X_dataRun1_HLT_frozen_v2"
0181 tag_name = "AlCaRecoTriggerBits_MuonDQM_v1_hlt"
0182 gt_map = self.connection.global_tag_map(global_tag_name=self.global_tag_name, tag_name=tag_name)
0183 self.assertTrue(isinstance(gt_map, self.connection.models["globaltagmap"]))
0184
0185 def test_get_empty_global_tag_map(self):
0186 empty_gt_map = self.connection.global_tag_map()
0187 self.assertTrue(isinstance(empty_gt_map, self.connection.models["globaltagmap"]))
0188 self.assertTrue(empty_gt_map.empty)
0189
0190 class tag_tests(querying_tests):
0191
0192 """
0193 Tags
0194 """
0195
0196 def test_get_tag(self):
0197
0198 tag_name = "EBAlignment_measured_v01_express"
0199 tag = self.connection.tag(name=tag_name)
0200
0201
0202 self.assertTrue(tag.name != None)
0203 self.assertTrue(isinstance(tag, self.connection.models["tag"]))
0204
0205 def test_get_empty_tag(self):
0206 tag = self.connection.tag()
0207 self.assertTrue(isinstance(tag, self.connection.models["tag"]))
0208 self.assertTrue(tag.empty)
0209
0210 def test_get_parent_global_tags(self):
0211 tag_name = "EBAlignment_measured_v01_express"
0212 tag = self.connection.tag(name=tag_name)
0213 self.assertTrue(tag != None)
0214 parent_global_tags = tag.parent_global_tags()
0215 self.assertTrue(parent_global_tags != None)
0216
0217 def test_all_tags_empty_tag(self):
0218 empty_tag = self.connection.tag()
0219 all_tags = empty_tag.all(amount=10)
0220 self.assertTrue(all_tags != None)
0221
0222 self.assertTrue(len(all_tags.data()) != 0)
0223
0224 def test_all_tags_non_empty_tag(self):
0225 tag_name = "EBAlignment_measured_v01_express"
0226 empty_tag = self.connection.tag(name=tag_name)
0227 all_tags = empty_tag.all(amount=10)
0228 self.assertTrue(all_tags != None)
0229
0230 self.assertTrue(len(all_tags.data()) != 0)
0231
0232 class iov_tests(querying_tests):
0233
0234 """
0235 IOVs
0236 """
0237
0238 def test_get_iov(self):
0239 tag_name = "EBAlignment_measured_v01_express"
0240 tag = self.connection.tag(name=tag_name)
0241 iovs = tag.iovs()
0242 self.assertTrue(isinstance(iovs, data_sources.json_list))
0243 raw_list = iovs.data()
0244 self.assertTrue(isinstance(raw_list, list))
0245 first_iov = raw_list[0]
0246 self.assertTrue(isinstance(first_iov, self.connection.models["iov"]))
0247
0248 def test_get_iovs_by_iov_query(self):
0249 tag_name = "EBAlignment_measured_v01_express"
0250 iovs = self.connection.iov(tag_name=tag_name)
0251 self.assertTrue(isinstance(iovs, data_sources.json_list))
0252 raw_list = iovs.data()
0253 self.assertTrue(isinstance(raw_list, list))
0254 first_iov = raw_list[0]
0255 self.assertTrue(isinstance(first_iov, self.connection.models["iov"]))
0256
0257 def test_get_empty_iov(self):
0258 empty_iov = self.connection.iov()
0259 self.assertTrue(isinstance(empty_iov, self.connection.models["iov"]))
0260 self.assertTrue(empty_iov.empty)
0261
0262 class payload_tests(querying_tests):
0263
0264 """
0265 Payloads
0266 """
0267
0268 def test_get_payload(self):
0269
0270 payload_hash = "00172cd62d8abae41915978d815ae62cc08ad8b9"
0271 payload = self.connection.payload(hash=payload_hash)
0272 self.assertTrue(isinstance(payload, self.connection.models["payload"]))
0273
0274 def test_get_empty_payload(self):
0275 payload = self.connection.payload()
0276 self.assertTrue(isinstance(payload, self.connection.models["payload"]))
0277 self.assertTrue(payload.empty)
0278
0279 def test_get_parent_tags_payload(self):
0280 payload_hash = "00172cd62d8abae41915978d815ae62cc08ad8b9"
0281 payload = self.connection.payload(hash=payload_hash)
0282 self.assertTrue(payload != None)
0283 parent_tags = payload.parent_tags()
0284 self.assertTrue(parent_tags != None)
0285
0286 def test_type_parent_tags(self):
0287 payload_hash = "00172cd62d8abae41915978d815ae62cc08ad8b9"
0288 payload = self.connection.payload(hash=payload_hash)
0289 self.assertTrue(payload != None)
0290 parent_tags = payload.parent_tags()
0291 self.assertTrue(isinstance(parent_tags, data_sources.json_list))
0292
0293 class misc_tests(querying_tests):
0294
0295 """
0296 Misc
0297 """
0298
0299 def test_search_everything(self):
0300 string_for_non_empty_result = "ecal"
0301 data = self.connection.search_everything(string_for_non_empty_result)
0302 self.assertTrue(len(data.get("global_tags").data()) != 0)
0303 self.assertTrue(len(data.get("tags").data()) != 0)
0304 self.assertTrue(len(data.get("payloads").data()) != 0)
0305
0306 class result_type_tests(querying_tests):
0307
0308 """
0309 Results types
0310 """
0311
0312 def test_factory_multiple_results(self):
0313 tags = self.connection.tag(time_type="Run")
0314 self.assertTrue(len(tags.data()) > 1)
0315
0316 def test_factory_empty_result(self):
0317 tag = self.connection.tag()
0318 self.assertTrue(tag.empty)
0319
0320 def test_factory_no_result(self):
0321 tag = self.connection.tag(name="dfasdf")
0322 self.assertTrue(tag == None)
0323
0324 def tearDown(self):
0325 self.connection.tear_down()