CondDBFW/tests/querying_tests.py

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         # hard coded global tag for now
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         # now check that the timestamp encoded in the strings is the same as the
0108         # datetime object that were converted
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         # get global tag maps
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         # this global tag is old, so this test is unlikely to fail since
0135         # its global tag maps will not fail
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         # verify both methods of difference
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         # hard code tag for now
0198         tag_name = "EBAlignment_measured_v01_express"
0199         tag = self.connection.tag(name=tag_name)
0200         # this tag exists, so shouldn't be NoneType
0201         # we gave the name, so that should at least be in the tag object
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         # there are always tags in the db
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         # there are always tags in the db
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         # hard coded payload for now
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()