Back to home page

Project CMSSW displayed by LXR

 
 

    


File indexing completed on 2021-02-14 12:50:48

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