CMS 3D CMS Logo

data_sources_tests.py
Go to the documentation of this file.
1 import unittest
2 import sys
3 import datetime
4 import pprint
5 import subprocess
6 import os
7 
8 import CondCore.Utilities.CondDBFW.querying as querying
9 import CondCore.Utilities.CondDBFW.data_sources as data_sources
10 import CondCore.Utilities.CondDBFW.data_formats as data_formats
11 import CondCore.Utilities.CondDBFW.shell as shell
12 import CondCore.Utilities.CondDBFW.models as models
13 
14 prod_connection_string = "frontier://FrontierProd/CMS_CONDITIONS"
15 secrets_source = None
16 
17 class data_sources_tests(unittest.TestCase):
18 
19  def setUp(self):
20  self.connection = querying.connect(prod_connection_string, secrets=secrets_source)
21 
23  test_list = list(range(0, 10))
24  json_list_object = data_sources.json_data_node.make(test_list)
25  self.assertTrue(json_list_object != None)
26  self.assertEqual(json_list_object.data(), test_list)
27  for n in range(0, len(test_list)):
28  self.assertEqual(json_list_object.get(n).data(), test_list[n])
29 
31  test_dict = {"key1" : "value1", "key2" : "value2", "key3" : "value3"}
32  json_dict_object = data_sources.json_data_node.make(test_dict)
33  self.assertTrue(json_dict_object != None)
34  self.assertEqual(json_dict_object.data(), test_dict)
35  for key in test_dict:
36  self.assertEqual(json_dict_object.get(key).data(), test_dict[key])
37 
39  structure = {"key1" : [{"a" : 1, "b" : 3}, {"a" : 4, "b" : 8}], "key2" : ["header1", "header2", "header3"]}
40  json_structure_object = data_sources.json_data_node.make(structure)
41  self.assertEqual(json_structure_object.get("key1").data(), structure["key1"])
42  self.assertEqual(json_structure_object.get("key2").data(), structure["key2"])
43 
44  def test_json_building(self):
45  structure = {"key1" : [{"a" : 1, "b" : 3}, {"a" : 4, "b" : 8}], "key2" : ["header1", "header2", "header3"]}
46  new_structure = data_sources.json_data_node.make({})
47  new_structure.add_key([], "key1")
48  new_structure.get("key1").add_child({"a" : 1, "b" : 3})
49  new_structure.get("key1").add_child({"a" : 4, "b" : 8})
50  new_structure.add_key([], "key2")
51  new_structure.get("key2").add_child("header1")
52  new_structure.get("key2").add_child("header2")
53  new_structure.get("key2").add_child("header3")
54  self.assertEqual(new_structure.data(), structure)
55 
56  def test_check_types(self):
57  test_list = list(range(0, 10))
58  test_dict = {"key1" : "value1", "key2" : "value2", "key3" : "value3"}
59  json_list_object = data_sources.json_data_node.make(test_list)
60  json_dict_object = data_sources.json_data_node.make(test_dict)
61  self.assertTrue(isinstance(json_list_object, data_sources.json_list))
62  self.assertTrue(isinstance(json_dict_object, data_sources.json_dict))
63 
64  def test_type_all_tags(self):
65  all_tags = self.connection.tag().all(amount=10)
66  self.assertTrue(isinstance(all_tags, data_sources.json_list))
67 
68  def test_type_all_iovs(self):
69  all_iovs = self.connection.iov().all(amount=10)
70  self.assertTrue(isinstance(all_iovs, data_sources.json_list))
71 
73  tag_name = "EBAlignment_measured_v01_express"
74  tag = self.connection.tag(name=tag_name)
75  self.assertTrue(tag != None)
76  parent_gts = tag.parent_global_tags()
77  self.assertTrue(isinstance(parent_gts, data_sources.json_list))
78 
79  def tearDown(self):
80  self.connection.tear_down()
def all(container)
workaround iterator generators for ROOT classes
Definition: cmstools.py:25
def connect(connection_data, mode="r", map_blobs=False, secrets=None, pooling=True)
Definition: querying.py:453
char data[epos_bytes_allocation]
Definition: EPOS_Wrapper.h:80