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         install_date = self.now()
 175         self.setup_maintainers()
 176         self.setup_suites()
 177         self.setup_poolfiles()
 178         self.source = {}
 179         self.source['hello_2.2-2'] = DBSource(source = 'hello', version = '2.2-2', \
 180             maintainer = self.maintainer['maintainer'], \
 181             changedby = self.maintainer['uploader'], \
 182             poolfile = self.file['hello_2.2-2.dsc'], install_date = install_date)
 183         self.source['hello_2.2-2'].suites.append(self.suite['sid'])
 184         self.source['hello_2.2-1'] = DBSource(source = 'hello', version = '2.2-1', \
 185             maintainer = self.maintainer['maintainer'], \
 186             changedby = self.maintainer['uploader'], \
 187             poolfile = self.file['hello_2.2-1.dsc'], install_date = install_date)
 188         self.source['hello_2.2-1'].suites.append(self.suite['sid'])
 189         self.source['gnome-hello_3.0-1'] = DBSource(source = 'gnome-hello', \
 190             version = '3.0-1', maintainer = self.maintainer['maintainer'], \
 191             changedby = self.maintainer['uploader'], \
 192             poolfile = self.file['gnome-hello_3.0-1.dsc'], install_date = install_date)
 193         self.source['gnome-hello_3.0-1'].suites.append(self.suite['sid'])
 194         self.source['sl_3.03-16'] = DBSource(source = 'sl', version = '3.03-16', \
 195             maintainer = self.maintainer['maintainer'], \
 196             changedby = self.maintainer['uploader'], \
 197             poolfile = self.file['sl_3.03-16.dsc'], install_date = install_date)
 198         self.source['sl_3.03-16'].suites.append(self.suite['squeeze'])
 199         self.source['sl_3.03-16'].suites.append(self.suite['sid'])
 200         self.session.add_all(self.source.values())
 201
 202     def setup_binaries(self):
 203         'create DBBinary objects'
 204
 205         if 'binary' in self.__dict__:
 206             return
 207         self.setup_sources()
 208         self.setup_architectures()
 209         self.binary = {}
 210         self.binary['hello_2.2-1_i386'] = DBBinary(package = 'hello', \
 211             source = self.source['hello_2.2-1'], version = '2.2-1', \
 212             maintainer = self.maintainer['maintainer'], \
 213             architecture = self.arch['i386'], \
 214             poolfile = self.file['hello_2.2-1_i386.deb'])
 215         self.binary['hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 216         self.binary['hello_2.2-1_i386'].suites.append(self.suite['sid'])
 217         self.binary['gnome-hello_2.2-1_i386'] = DBBinary(package = 'gnome-hello', \
 218             source = self.source['hello_2.2-1'], version = '2.2-1', \
 219             maintainer = self.maintainer['maintainer'], \
 220             architecture = self.arch['i386'], \
 221             poolfile = self.file['gnome-hello_2.2-1_i386.deb'])
 222         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 223         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['sid'])
 224         self.binary['gnome-hello_3.0-1_i386'] = DBBinary(package = 'gnome-hello', \
 225             source = self.source['gnome-hello_3.0-1'], version = '3.0-1', \
 226             maintainer = self.maintainer['maintainer'], \
 227             architecture = self.arch['i386'], \
 228             poolfile = self.file['gnome-hello_3.0-1_i386.deb'])
 229         self.binary['gnome-hello_3.0-1_i386'].suites.append(self.suite['sid'])
 230         self.binary['python-hello_2.2-1_i386'] = DBBinary(package = 'python-hello', \
 231             source = self.source['hello_2.2-1'], version = '2.2-1', \
 232             maintainer = self.maintainer['maintainer'], \
 233             architecture = self.arch['all'], \
 234             poolfile = self.file['python-hello_2.2-1_all.deb'])
 235         self.binary['python-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 236         self.session.add_all(self.binary.values())
 237
 238     def setup_overridetypes(self):
 239         '''
 240         Setup self.otype of class OverrideType.
 241         '''
 242         if 'otype' in self.__dict__:
 243             return
 244         self.otype = {}
 245         self.otype['deb'] = OverrideType(overridetype = 'deb')
 246         self.session.add_all(self.otype.values())
 247         self.session.flush()
 248
 249     def setup_sections(self):
 250         '''
 251         Setup self.section of class Section.
 252         '''
 253         if 'section' in self.__dict__:
 254             return
 255         self.section = {}
 256         self.section['python'] = Section(section = 'python')
 257         self.session.add_all(self.section.values())
 258         self.session.flush()
 259
 260     def setup_priorities(self):
 261         '''
 262         Setup self.prio of class Priority.
 263         '''
 264         if 'prio' in self.__dict__:
 265             return
 266         self.prio = {}
 267         self.prio['standard'] = Priority(priority = 'standard', level = 7)
 268         self.session.add_all(self.prio.values())
 269         self.session.flush()
 270
 271     def setUp(self):
 272         if self.metadata is None:
 273             self.initialize()
 274         self.session = DBConn().session()
 275
 276     def now(self):
 277         """
 278         Returns the current time at the db server. Please note the function
 279         returns the same value as long as it is in the same transaction. You
 280         should self.session.rollback() (or commit) if you rely on getting a
 281         fresh timestamp.
 282         """
 283
 284         return self.session.query(func.now()).scalar()
 285
 286     def classes_to_clean(self):
 287         """
 288         The function classes_to_clean() returns a list of classes. All objects
 289         of each class will be deleted from the database in tearDown(). This
 290         function should be overridden in derived test cases as needed.
 291         """
 292         return ()
 293
 294     def tearDown(self):
 295         self.session.rollback()
 296         for class_ in self.classes_to_clean():
 297             self.session.query(class_).delete()
 298         self.session.commit()
 299         # usually there is no need to drop all tables here
 300         #self.metadata.drop_all()
 301