File indexing completed on 2024-04-06 12:01:53
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 prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"
0015 secrets_source = None
0016
0017 class data_sources_tests(unittest.TestCase):
0018
0019 def setUp(self):
0020 self.connection = querying.connect(prod_connection_string, secrets=secrets_source)
0021
0022 def test_make_json_list(self):
0023 test_list = list(range(0, 10))
0024 json_list_object = data_sources.json_data_node.make(test_list)
0025 self.assertTrue(json_list_object != None)
0026 self.assertEqual(json_list_object.data(), test_list)
0027 for n in range(0, len(test_list)):
0028 self.assertEqual(json_list_object.get(n).data(), test_list[n])
0029
0030 def test_make_json_dict(self):
0031 test_dict = {"key1" : "value1", "key2" : "value2", "key3" : "value3"}
0032 json_dict_object = data_sources.json_data_node.make(test_dict)
0033 self.assertTrue(json_dict_object != None)
0034 self.assertEqual(json_dict_object.data(), test_dict)
0035 for key in test_dict:
0036 self.assertEqual(json_dict_object.get(key).data(), test_dict[key])
0037
0038 def test_json_navigation(self):
0039 structure = {"key1" : [{"a" : 1, "b" : 3}, {"a" : 4, "b" : 8}], "key2" : ["header1", "header2", "header3"]}
0040 json_structure_object = data_sources.json_data_node.make(structure)
0041 self.assertEqual(json_structure_object.get("key1").data(), structure["key1"])
0042 self.assertEqual(json_structure_object.get("key2").data(), structure["key2"])
0043
0044 def test_json_building(self):
0045 structure = {"key1" : [{"a" : 1, "b" : 3}, {"a" : 4, "b" : 8}], "key2" : ["header1", "header2", "header3"]}
0046 new_structure = data_sources.json_data_node.make({})
0047 new_structure.add_key([], "key1")
0048 new_structure.get("key1").add_child({"a" : 1, "b" : 3})
0049 new_structure.get("key1").add_child({"a" : 4, "b" : 8})
0050 new_structure.add_key([], "key2")
0051 new_structure.get("key2").add_child("header1")
0052 new_structure.get("key2").add_child("header2")
0053 new_structure.get("key2").add_child("header3")
0054 self.assertEqual(new_structure.data(), structure)
0055
0056 def test_check_types(self):
0057 test_list = list(range(0, 10))
0058 test_dict = {"key1" : "value1", "key2" : "value2", "key3" : "value3"}
0059 json_list_object = data_sources.json_data_node.make(test_list)
0060 json_dict_object = data_sources.json_data_node.make(test_dict)
0061 self.assertTrue(isinstance(json_list_object, data_sources.json_list))
0062 self.assertTrue(isinstance(json_dict_object, data_sources.json_dict))
0063
0064 def test_type_all_tags(self):
0065 all_tags = self.connection.tag().all(amount=10)
0066 self.assertTrue(isinstance(all_tags, data_sources.json_list))
0067
0068 def test_type_all_iovs(self):
0069 all_iovs = self.connection.iov().all(amount=10)
0070 self.assertTrue(isinstance(all_iovs, data_sources.json_list))
0071
0072 def test_type_parent_global_tags(self):
0073 tag_name = "EBAlignment_measured_v01_express"
0074 tag = self.connection.tag(name=tag_name)
0075 self.assertTrue(tag != None)
0076 parent_gts = tag.parent_global_tags()
0077 self.assertTrue(isinstance(parent_gts, data_sources.json_list))
0078
0079 def tearDown(self):
0080 self.connection.tear_down()
