CondDBFW/tests/querying_tests.py

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