1 from base_test import DakTestCase, fixture
3 from daklib.config import Config
4 from daklib.dbconn import DBConn
6 from sqlalchemy import create_engine, func, __version__
7 from sqlalchemy.exc import SADeprecationWarning
8 from sqlalchemy.schema import DDL
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']
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()"""
32 class DBDakTestCase(DakTestCase):
33 def execute(self, statement):
34 DDL(statement).execute(self.metadata.bind)
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))
46 if cnf["DB::Name"] in ('backports', 'obscurity', 'projectb'):
47 self.fail("You have configured an invalid database name: '%s'." % \
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"]
57 connstr = "postgres:///%s" % cnf["DB::Name"]
58 if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
59 connstr += "?port=%s" % cnf["DB::Port"]
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)
66 self.metadata.bind = create_engine(connstr)
67 self.metadata.create_all()
68 self.create_all_triggers()
71 if self.metadata is None:
73 self.session = DBConn().session()
76 "returns the current time at the db server"
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()
84 def classes_to_clean(self):
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.
93 self.session.rollback()
94 for class_ in self.classes_to_clean():
95 self.session.query(class_).delete()
97 # usually there is no need to drop all tables here
98 #self.metadata.drop_all()