tests/db_test.py

   1 from base_test import DakTestCase, fixture
   2
   3 from daklib.config import Config
   4 from daklib.dbconn import *
   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_suites(self):
  71         "setup a hash of Suite objects in self.suite"
  72
  73         if 'suite' in self.__dict__:
  74             return
  75         self.suite = {}
  76         for suite_name in ('lenny', 'squeeze', 'sid'):
  77             self.suite[suite_name] = Suite(suite_name = suite_name, version = '-')
  78         self.session.add_all(self.suite.values())
  79
  80     def setup_architectures(self):
  81         "setup Architecture objects in self.arch and connect to suites"
  82
  83         if 'arch' in self.__dict__:
  84             return
  85         self.setup_suites()
  86         self.arch = {}
  87         for arch_string in ('source', 'all', 'i386', 'amd64', 'kfreebsd-i386'):
  88             self.arch[arch_string] = Architecture(arch_string)
  89             if arch_string != 'kfreebsd-i386':
  90                 self.arch[arch_string].suites = self.suite.values()
  91             else:
  92                 self.arch[arch_string].suites = [self.suite['squeeze'], self.suite['sid']]
  93         # hard code ids for source and all
  94         self.arch['source'].arch_id = 1
  95         self.arch['all'].arch_id = 2
  96         self.session.add_all(self.arch.values())
  97
  98     def setup_components(self):
  99         'create some Component objects'
 100
 101         if 'comp' in self.__dict__:
 102             return
 103         self.comp = {}
 104         self.comp['main'] = Component(component_name = 'main')
 105         self.comp['contrib'] = Component(component_name = 'contrib')
 106         self.session.add_all(self.comp.values())
 107
 108     def setup_locations(self):
 109         'create some Location objects'
 110
 111         if 'loc' in self.__dict__:
 112             return
 113         self.setup_components()
 114         self.loc = {}
 115         self.loc['main'] = Location( \
 116             path = '/srv/ftp-master.debian.org/ftp/pool/', \
 117             component = self.comp['main'])
 118         self.loc['contrib'] = Location( \
 119             path = '/srv/ftp-master.debian.org/ftp/pool/', \
 120             component = self.comp['contrib'])
 121         self.session.add_all(self.loc.values())
 122
 123     def setup_poolfiles(self):
 124         'create some PoolFile objects'
 125
 126         if 'file' in self.__dict__:
 127             return
 128         self.setup_locations()
 129         self.file = {}
 130         self.file['hello_2.2-3.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-3.dsc', \
 131             location = self.loc['main'], filesize = 0, md5sum = '')
 132         self.file['hello_2.2-2.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-2.dsc', \
 133             location = self.loc['main'], filesize = 0, md5sum = '')
 134         self.file['hello_2.2-1.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-1.dsc', \
 135             location = self.loc['main'], filesize = 0, md5sum = '')
 136         self.file['gnome-hello_3.0-1.dsc'] = PoolFile( \
 137             filename = 'main/g/gnome-hello/gnome-hello_3.0-1.dsc', \
 138             location = self.loc['contrib'], filesize = 0, md5sum = '')
 139         self.file['hello_2.2-1_i386.deb'] = PoolFile( \
 140             filename = 'main/h/hello/hello_2.2-1_i386.deb', \
 141             location = self.loc['main'], filesize = 0, md5sum = '')
 142         self.file['gnome-hello_2.2-1_i386.deb'] = PoolFile( \
 143             filename = 'main/h/hello/gnome-hello_2.2-1_i386.deb', \
 144             location = self.loc['main'], filesize = 0, md5sum = '')
 145         self.file['python-hello_2.2-1_all.deb'] = PoolFile( \
 146             filename = 'main/h/hello/python-hello_2.2-1_all.deb', \
 147             location = self.loc['main'], filesize = 0, md5sum = '')
 148         self.file['gnome-hello_3.0-1_i386.deb'] = PoolFile( \
 149             filename = 'main/g/gnome-hello/gnome-hello_3.0-1_i386.deb', \
 150             location = self.loc['contrib'], filesize = 0, md5sum = '')
 151         self.file['sl_3.03-16.dsc'] = PoolFile(filename = 'main/s/sl/sl_3.03-16.dsc', \
 152             location = self.loc['main'], filesize = 0, md5sum = '')
 153         self.file['python2.6_2.6.6-8.dsc'] = PoolFile( \
 154             filename = 'main/p/python2.6/python2.6_2.6.6-8.dsc', \
 155             location = self.loc['main'], filesize = 0, md5sum = '')
 156         self.session.add_all(self.file.values())
 157
 158     def setup_maintainers(self):
 159         'create some Maintainer objects'
 160
 161         if 'maintainer' in self.__dict__:
 162             return
 163         self.maintainer = {}
 164         self.maintainer['maintainer'] = Maintainer(name = 'Mr. Maintainer')
 165         self.maintainer['uploader'] = Maintainer(name = 'Mrs. Uploader')
 166         self.maintainer['lazyguy'] = Maintainer(name = 'Lazy Guy')
 167         self.session.add_all(self.maintainer.values())
 168
 169     def setup_sources(self):
 170         'create DBSource objects'
 171
 172         if 'source' in self.__dict__:
 173             return
 174         self.setup_maintainers()
 175         self.setup_suites()
 176         self.setup_poolfiles()
 177         self.source = {}
 178         self.source['hello_2.2-2'] = DBSource(source = 'hello', version = '2.2-2', \
 179             maintainer = self.maintainer['maintainer'], \
 180             changedby = self.maintainer['uploader'], \
 181             poolfile = self.file['hello_2.2-2.dsc'], install_date = self.now())
 182         self.source['hello_2.2-2'].suites.append(self.suite['sid'])
 183         self.source['hello_2.2-1'] = DBSource(source = 'hello', version = '2.2-1', \
 184             maintainer = self.maintainer['maintainer'], \
 185             changedby = self.maintainer['uploader'], \
 186             poolfile = self.file['hello_2.2-1.dsc'], install_date = self.now())
 187         self.source['hello_2.2-1'].suites.append(self.suite['sid'])
 188         self.source['gnome-hello_3.0-1'] = DBSource(source = 'gnome-hello', \
 189             version = '3.0-1', maintainer = self.maintainer['maintainer'], \
 190             changedby = self.maintainer['uploader'], \
 191             poolfile = self.file['gnome-hello_3.0-1.dsc'], install_date = self.now())
 192         self.source['gnome-hello_3.0-1'].suites.append(self.suite['sid'])
 193         self.source['sl_3.03-16'] = DBSource(source = 'sl', version = '3.03-16', \
 194             maintainer = self.maintainer['maintainer'], \
 195             changedby = self.maintainer['uploader'], \
 196             poolfile = self.file['sl_3.03-16.dsc'], install_date = self.now())
 197         self.source['sl_3.03-16'].suites.append(self.suite['squeeze'])
 198         self.source['sl_3.03-16'].suites.append(self.suite['sid'])
 199         self.session.add_all(self.source.values())
 200
 201     def setup_binaries(self):
 202         'create DBBinary objects'
 203
 204         if 'binary' in self.__dict__:
 205             return
 206         self.setup_sources()
 207         self.setup_architectures()
 208         self.binary = {}
 209         self.binary['hello_2.2-1_i386'] = DBBinary(package = 'hello', \
 210             source = self.source['hello_2.2-1'], version = '2.2-1', \
 211             maintainer = self.maintainer['maintainer'], \
 212             architecture = self.arch['i386'], \
 213             poolfile = self.file['hello_2.2-1_i386.deb'])
 214         self.binary['hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 215         self.binary['hello_2.2-1_i386'].suites.append(self.suite['sid'])
 216         self.binary['gnome-hello_2.2-1_i386'] = DBBinary(package = 'gnome-hello', \
 217             source = self.source['hello_2.2-1'], version = '2.2-1', \
 218             maintainer = self.maintainer['maintainer'], \
 219             architecture = self.arch['i386'], \
 220             poolfile = self.file['gnome-hello_2.2-1_i386.deb'])
 221         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 222         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['sid'])
 223         self.binary['gnome-hello_3.0-1_i386'] = DBBinary(package = 'gnome-hello', \
 224             source = self.source['gnome-hello_3.0-1'], version = '3.0-1', \
 225             maintainer = self.maintainer['maintainer'], \
 226             architecture = self.arch['i386'], \
 227             poolfile = self.file['gnome-hello_3.0-1_i386.deb'])
 228         self.binary['gnome-hello_3.0-1_i386'].suites.append(self.suite['sid'])
 229         self.binary['python-hello_2.2-1_i386'] = DBBinary(package = 'python-hello', \
 230             source = self.source['hello_2.2-1'], version = '2.2-1', \
 231             maintainer = self.maintainer['maintainer'], \
 232             architecture = self.arch['all'], \
 233             poolfile = self.file['python-hello_2.2-1_all.deb'])
 234         self.binary['python-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 235         self.session.add_all(self.binary.values())
 236
 237     def setUp(self):
 238         if self.metadata is None:
 239             self.initialize()
 240         self.session = DBConn().session()
 241
 242     def now(self):
 243         "returns the current time at the db server"
 244
 245         # we fetch a fresh session each time to avoid caching
 246         local_session = DBConn().session()
 247         current_time = local_session.query(func.now()).scalar()
 248         local_session.close()
 249         return current_time
 250
 251     def classes_to_clean(self):
 252         """
 253         The function classes_to_clean() returns a list of classes. All objects
 254         of each class will be deleted from the database in tearDown(). This
 255         function should be overridden in derived test cases as needed.
 256         """
 257         return ()
 258
 259     def tearDown(self):
 260         self.session.rollback()
 261         for class_ in self.classes_to_clean():
 262             self.session.query(class_).delete()
 263         self.session.commit()
 264         # usually there is no need to drop all tables here
 265         #self.metadata.drop_all()
 266