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 prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"
0015 secrets_source = None
0016
0017 class data_formats_tests(unittest.TestCase):
0018
0019 def setUp(self):
0020 self.connection = querying.connect(prod_connection_string, secrets=secrets_source)
0021
0022 def test_orm_objects_to_dicts(self):
0023 tags = self.connection.tag().all(amount=10)
0024 list_of_dicts = data_formats._objects_to_dicts(tags)
0025 self.assertEqual(list_of_dicts.__class__.__name__, "json_list")
0026 for tag in tags:
0027 self.assertTrue(isinstance(tag, self.connection.models["tag"]))
0028
0029 def test_dicts_to_orm_objects(self):
0030 models_to_test = map(self.connection.model, ["global_tag", "tag", "payload", "iov"])
0031 for model in models_to_test:
0032 model_name = models.class_name_to_column(model).lower()
0033 objects = getattr(self.connection, model_name)().all(amount=5)
0034 dicts = data_formats._objects_to_dicts(objects)
0035 orm_objects = data_formats._dicts_to_orm_objects(self.connection.model(model_name), dicts)
0036 self.assertTrue(dicts != None)
0037 for obj in orm_objects:
0038 self.assertEqual(models.class_name_to_column(obj.__class__).lower(), model_name)
0039 headers = model.headers
0040 for header in headers:
0041 try:
0042 test = getattr(obj, header)
0043 header_exists = True
0044 except:
0045 print("'%s' doesn't exist." % header)
0046 header_exists = False
0047 self.assertTrue(header_exists)
0048
0049 def tearDown(self):
0050 self.connection.tear_down()
