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 (querying_tests)
def	global_tag_tests (querying_tests)

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

Function Documentation

factory_tests()

def querying_tests.factory_tests

(

querying_tests

)

Factory

Definition at line 30 of file querying_tests.py.

def factory_tests(querying_tests):

    """
    Factory
    """

    def test_check_keys_in_models(self):
        """
        Verifies that each key that is required is present in self.connection's model dictionary,
        and also that self.connection has a proxy method for each model that is generated.
        """
        keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
        for key in keys:
            self.assertTrue(key in self.connection.models.keys())
        proxy_method_names = ["global_tag", "global_tag_map", "tag", "iov", "payload"]
        for name in proxy_method_names:
            self.assertTrue(hasattr(self.connection, name))

    def test_raw_factory_calls_all_models(self):
        """
        Verifies that the factory object held by the connection generates classes belonging to the type
        held by self.connection.
        """
        keys = ["globaltag", "globaltagmap", "tag", "iov", "payload"]
        for key in keys:
            self.assertTrue(isinstance(self.connection.factory.object(key), self.connection.models[key]))

global_tag_tests()

def querying_tests.global_tag_tests

(

querying_tests

)

Global Tags

Definition at line 57 of file querying_tests.py.

References data, print(), isotrackApplyRegressor.range, and utils.to_datetime().

def global_tag_tests(querying_tests):

    """
    Global Tags
    """

    def test_get_global_tag(self):
        # hard coded global tag for now
        global_tag = self.connection.global_tag(name=self.global_tag_name)
        self.assertTrue(global_tag.name != None)
        self.assertTrue(global_tag.tags() != None)
        self.assertTrue(isinstance(global_tag, self.connection.models["globaltag"]))

    def test_get_empty_global_tag(self):
        empty_gt = self.connection.global_tag()
        self.assertTrue(isinstance(empty_gt, self.connection.models["globaltag"]))
        self.assertTrue(empty_gt.empty)

    def test_all_global_tags_empty_gt(self):
        empty_gt = self.connection.global_tag()
        all_gts = empty_gt.all(amount=10)
        self.assertTrue(all_gts != None)
        self.assertTrue(len(all_gts.data()) != 0)

    def test_all_method_parameters(self):
        if self.connection.connection_data["host"].lower() == "frontier":
            print("Cannot query for timestamps on frontier connection.")
            return
        empty_gt = self.connection.global_tag()
        sample_time = datetime.datetime(year=2016, month=1, day=1)
        now = datetime.datetime.now()
        time_range = Range(sample_time, now)
        time_radius = Radius(sample_time, datetime.timedelta(weeks=4))
        all_gts_in_interval = empty_gt.all(insertion_time=time_range).data()
        for gt in all_gts_in_interval:
            self.assertTrue(sample_time <= gt.insertion_time <= now)

    def test_as_dicts_method_without_timestamps_conversion(self):
        global_tag = self.connection.global_tag(name=self.global_tag_name)
        dict_form = global_tag.as_dicts(convert_timestamps=False)
        self.assertTrue(isinstance(dict_form, dict))
        self.assertTrue(dict_form["insertion_time"], datetime.datetime)
        self.assertTrue(dict_form["snapshot_time"], datetime.datetime)

    def test_as_dicts_method_with_timestamps_conversion(self):
        global_tag = self.connection.global_tag(name=self.global_tag_name)
        dict_form = global_tag.as_dicts(convert_timestamps=True)
        self.assertTrue(isinstance(dict_form, dict))
        self.assertTrue(dict_form["insertion_time"], str)
        self.assertTrue(dict_form["snapshot_time"], str)

        # now check that the timestamp encoded in the strings is the same as the
        # datetime object that were converted
        self.assertTrue(to_datetime(dict_form["insertion_time"]) == global_tag.insertion_time)
        self.assertTrue(to_datetime(dict_form["snapshot_time"]) == global_tag.snapshot_time)

    def test_get_tag_maps_in_global_tag(self):
        global_tag = self.connection.global_tag(name=self.global_tag_name)

        # get global tag maps
        global_tag_maps = global_tag.tags()
        self.assertTrue(isinstance(global_tag_maps, data_sources.json_list))
        global_tag_maps_list = global_tag_maps.data()
        self.assertTrue(isinstance(global_tag_maps_list, list))
        self.assertTrue(len(global_tag_maps_list) != 0)

        for gt_map in global_tag_maps_list:
            self.assertTrue(isinstance(gt_map, self.connection.models["globaltagmap"]))

    def test_get_tag_maps_in_global_tag_with_parameters(self):
        global_tag = self.connection.global_tag(name=self.global_tag_name)

        global_tag_maps_specific_record = global_tag.tags(record="AlCaRecoTriggerBitsRcd")

        self.assertTrue(isinstance(global_tag_maps_specific_record, data_sources.json_list))

        gt_maps_spec_record_list = global_tag_maps_specific_record.data()
        self.assertTrue(isinstance(gt_maps_spec_record_list, list))
        # this global tag is old, so this test is unlikely to fail since
        # its global tag maps will not fail
        self.assertTrue(len(gt_maps_spec_record_list) == 3)

    def test_get_all_iovs_within_range(self):
        global_tag = self.connection.global_tag(name=self.global_tag_name)

        since_range = self.connection.range(200000, 300000)
        iovs = global_tag.iovs(since=since_range)
        self.assertTrue(isinstance(iovs, data_sources.json_list))
        iovs_list = iovs.data()
        self.assertTrue(isinstance(iovs_list, list))

        for iov in iovs_list:
            self.assertTrue(isinstance(iov, self.connection.models["iov"]))
            self.assertTrue(200000 <= iov.since <= 300000)

    def test_gt_diff(self):
        gt1 = self.connection.global_tag(name="74X_dataRun1_v1")
        gt2 = self.connection.global_tag(name="74X_dataRun1_v3")
        difference_by_arithmetic = gt1 - gt2
        difference_by_method = gt1.diff(gt2)

        # verify both methods of difference
        self.assertTrue(isinstance(difference_by_arithmetic, data_sources.json_list))
        self.assertTrue(isinstance(difference_by_method, data_sources.json_list))

        difference_arithmetic_list = difference_by_arithmetic.data()
        difference_method_list = difference_by_method.data()
        self.assertTrue(isinstance(difference_arithmetic_list, list))
        self.assertTrue(isinstance(difference_method_list, list))

        self.assertTrue(len(difference_arithmetic_list) == len(difference_method_list))

        for n in range(len(difference_arithmetic_list)):
            self.assertTrue(difference_arithmetic_list[n]["%s Tag" % gt1.name] != difference_arithmetic_list[n]["%s Tag" % gt2.name])
            self.assertTrue(difference_method_list[n]["%s Tag" % gt1.name] != difference_method_list[n]["%s Tag" % gt2.name])

Variable Documentation

prod_connection_string

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

Definition at line 18 of file querying_tests.py.

secrets_source

querying_tests.secrets_source = None

Definition at line 19 of file querying_tests.py.