CondDBFW/tests/payload_tests.py

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 payload_tests(unittest.TestCase):
0018
0019     def setUp(self):
0020         # set up a connection to oracle
0021         self.connection = querying.connect(prod_connection_string, map_blobs=True, secrets=secrets_source)
0022         # get a payload
0023         self.payload = self.connection.payload(hash="00172cd62d8abae41915978d815ae62cc08ad8b9")
0024         if not(os.path.isfile("test_suite.sqlite")):
0025             # create file
0026             handle = open("test_suite.sqlite", "w")
0027             handle.close()
0028         # insert schema
0029         if os.path.isfile("simple_conditions_schema.sql"):
0030             try:
0031                 process = subprocess.Popen("sqlite3 test_suite.sqlite < simple_conditions_schema.sql")
0032                 result = process.communicate()[0]
0033             except Exception as e:
0034                 self.test_write_blob_to_sqlite = unittest.skip("Can't setup sqlite database file.")(self.test_write_blob_to_sqlite)
0035
0036     def test_recomputed_hash(self):
0037         import hashlib
0038         recomputed_hash = hashlib.sha1(self.payload.object_type)
0039         recomputed_hash.update(self.payload.data)
0040         recomputed_hash = recomputed_hash.hexdigest()
0041         self.assertEqual(recomputed_hash, self.payload.hash)
0042
0043     def test_write_blob_to_sqlite(self):
0044         import os
0045         # open sqlite file in CondDBFW
0046         sqlite_con = querying.connect("sqlite://test_suite.sqlite", map_blobs=True)
0047         # write to sqlite file
0048         sqlite_con.write_and_commit(self.payload)
0049         # read payload from sqlite file, check for equality between blobs
0050         sqlite_payload = sqlite_con.payload(hash=self.payload.hash)
0051         self.assertEqual(sqlite_payload.data, self.payload.data)
0052         # delete payload from sqlite file
0053         tmp_sqlite_connection = sqlite_con.engine.connect()
0054         result = tmp_sqlite_connection.execute("delete from payload where hash=?", self.payload.hash)
0055         tmp_sqlite_connection.close()
0056
0057         # check that payload isn't in sqlite anymore
0058         payload_in_sqlite = sqlite_con.payload(hash=self.payload.hash)
0059         self.assertEqual(payload_in_sqlite, None)