tests/db_test.py

   1 from base_test import DakTestCase, fixture
   2
   3 from daklib.config import Config
   4 from daklib.dbconn import DBConn
   5
   6 from sqlalchemy import create_engine, func, __version__
   7 from sqlalchemy.exc import SADeprecationWarning
   8 from sqlalchemy.schema import DDL
   9
  10 import pickle
  11 import warnings
  12
  13 all_tables = ['architecture', 'archive', 'bin_associations', 'bin_contents',
  14     'binaries', 'binary_acl', 'binary_acl_map', 'build_queue', 'build_queue_files',
  15     'changes', 'changes_pending_binaries', 'changes_pending_files',
  16     'changes_pending_files_map', 'changes_pending_source',
  17     'changes_pending_source_files', 'changes_pool_files', 'component', 'config',
  18     'dsc_files', 'files', 'fingerprint', 'keyring_acl_map', 'keyrings', 'location',
  19     'maintainer', 'new_comments', 'override', 'override_type', 'policy_queue',
  20     'priority', 'section', 'source', 'source_acl', 'src_associations',
  21     'src_format', 'src_uploaders', 'suite', 'suite_architectures',
  22     'suite_build_queue_copy', 'suite_src_formats', 'uid', 'upload_blocks']
  23
  24 drop_plpgsql = "DROP LANGUAGE IF EXISTS plpgsql CASCADE"
  25 create_plpgsql = "CREATE LANGUAGE plpgsql"
  26 create_function = """CREATE OR REPLACE FUNCTION tfunc_set_modified() RETURNS trigger AS $$
  27     BEGIN NEW.modified = now(); return NEW; END;
  28     $$ LANGUAGE 'plpgsql'"""
  29 create_trigger = """CREATE TRIGGER modified_%s BEFORE UPDATE ON %s
  30     FOR EACH ROW EXECUTE PROCEDURE tfunc_set_modified()"""
  31
  32 class DBDakTestCase(DakTestCase):
  33     def execute(self, statement):
  34         DDL(statement).execute(self.metadata.bind)
  35
  36     def create_all_triggers(self):
  37         for statement in (drop_plpgsql, create_plpgsql, create_function):
  38             self.execute(statement)
  39         for table in all_tables:
  40             self.execute(create_trigger % (table, table))
  41
  42     metadata = None
  43
  44     def initialize(self):
  45         cnf = Config()
  46         if cnf["DB::Name"] in ('backports', 'obscurity', 'projectb'):
  47             self.fail("You have configured an invalid database name: '%s'." % \
  48                     cnf["DB::Name"])
  49         if cnf["DB::Host"]:
  50             # TCP/IP
  51             connstr = "postgres://%s" % cnf["DB::Host"]
  52             if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
  53                 connstr += ":%s" % cnf["DB::Port"]
  54             connstr += "/%s" % cnf["DB::Name"]
  55         else:
  56             # Unix Socket
  57             connstr = "postgres:///%s" % cnf["DB::Name"]
  58             if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
  59                 connstr += "?port=%s" % cnf["DB::Port"]
  60
  61         pickle_filename = 'db-metadata-%s.pkl' % __version__
  62         pickle_file = open(fixture(pickle_filename), 'r')
  63         DBDakTestCase.metadata = pickle.load(pickle_file)
  64         self.metadata.ddl_listeners = pickle.load(pickle_file)
  65         pickle_file.close()
  66         self.metadata.bind = create_engine(connstr)
  67         self.metadata.create_all()
  68         self.create_all_triggers()
  69
  70     def setUp(self):
  71         if self.metadata is None:
  72             self.initialize()
  73         self.session = DBConn().session()
  74
  75     def now(self):
  76         "returns the current time at the db server"
  77
  78         # we fetch a fresh session each time to avoid caching
  79         local_session = DBConn().session()
  80         current_time = local_session.query(func.now()).scalar()
  81         local_session.close()
  82         return current_time
  83
  84     def classes_to_clean(self):
  85         """
  86         The function classes_to_clean() returns a list of classes. All objects
  87         of each class will be deleted from the database in tearDown(). This
  88         function should be overridden in derived test cases as needed.
  89         """
  90         return ()
  91
  92     def tearDown(self):
  93         self.session.rollback()
  94         for class_ in self.classes_to_clean():
  95             self.session.query(class_).delete()
  96         self.session.commit()
  97         # usually there is no need to drop all tables here
  98         #self.metadata.drop_all()
  99