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         for type_ in ('deb', 'udeb'):
 246             self.otype[type_] = OverrideType(overridetype = type_)
 247         self.session.add_all(self.otype.values())
 248         self.session.flush()
 249
 250     def setup_sections(self):
 251         '''
 252         Setup self.section of class Section.
 253         '''
 254         if 'section' in self.__dict__:
 255             return
 256         self.section = {}
 257         self.section['python'] = Section(section = 'python')
 258         self.session.add_all(self.section.values())
 259         self.session.flush()
 260
 261     def setup_priorities(self):
 262         '''
 263         Setup self.prio of class Priority.
 264         '''
 265         if 'prio' in self.__dict__:
 266             return
 267         self.prio = {}
 268         self.prio['standard'] = Priority(priority = 'standard', level = 7)
 269         self.session.add_all(self.prio.values())
 270         self.session.flush()
 271
 272     def setup_overrides(self):
 273         '''
 274         Setup self.override of class Override.
 275         '''
 276         if 'override' in self.__dict__:
 277             return
 278         self.setup_suites()
 279         self.setup_components()
 280         self.setup_overridetypes()
 281         self.setup_sections()
 282         self.setup_priorities()
 283         self.override = {}
 284         self.override['hello_sid_main_udeb'] = Override(package = 'hello', \
 285             suite = self.suite['sid'], component = self.comp['main'], \
 286             overridetype = self.otype['udeb'], \
 287             section = self.section['python'], priority = self.prio['standard'])
 288         self.override['hello_squeeze_main_deb'] = Override(package = 'hello', \
 289             suite = self.suite['squeeze'], component = self.comp['main'], \
 290             overridetype = self.otype['deb'], \
 291             section = self.section['python'], priority = self.prio['standard'])
 292         self.override['hello_lenny_contrib_deb'] = Override(package = 'hello', \
 293             suite = self.suite['lenny'], component = self.comp['contrib'], \
 294             overridetype = self.otype['deb'], \
 295             section = self.section['python'], priority = self.prio['standard'])
 296         self.session.add_all(self.override.values())
 297         self.session.flush()
 298
 299     def setUp(self):
 300         if self.metadata is None:
 301             self.initialize()
 302         self.session = DBConn().session()
 303
 304     def now(self):
 305         """
 306         Returns the current time at the db server. Please note the function
 307         returns the same value as long as it is in the same transaction. You
 308         should self.session.rollback() (or commit) if you rely on getting a
 309         fresh timestamp.
 310         """
 311
 312         return self.session.query(func.now()).scalar()
 313
 314     def classes_to_clean(self):
 315         """
 316         The function classes_to_clean() returns a list of classes. All objects
 317         of each class will be deleted from the database in tearDown(). This
 318         function should be overridden in derived test cases as needed.
 319         """
 320         return ()
 321
 322     def tearDown(self):
 323         self.session.rollback()
 324         for class_ in self.classes_to_clean():
 325             self.session.query(class_).delete()
 326         self.session.commit()
 327         # usually there is no need to drop all tables here
 328         #self.metadata.drop_all()
 329