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         for name in ('main', 'contrib', 'non-free'):
 105             self.comp[name] = Component(component_name = name)
 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 = fixture('ftp/pool/'), component = self.comp['main'])
 117         self.loc['contrib'] = Location( \
 118             path = fixture('ftp/pool/'), component = self.comp['contrib'])
 119         self.session.add_all(self.loc.values())
 120
 121     def setup_poolfiles(self):
 122         'create some PoolFile objects'
 123
 124         if 'file' in self.__dict__:
 125             return
 126         self.setup_locations()
 127         self.file = {}
 128         self.file['hello_2.2-3.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-3.dsc', \
 129             location = self.loc['main'], filesize = 0, md5sum = '')
 130         self.file['hello_2.2-2.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-2.dsc', \
 131             location = self.loc['main'], filesize = 0, md5sum = '')
 132         self.file['hello_2.2-1.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-1.dsc', \
 133             location = self.loc['main'], filesize = 0, md5sum = '')
 134         self.file['gnome-hello_3.0-1.dsc'] = PoolFile( \
 135             filename = 'main/g/gnome-hello/gnome-hello_3.0-1.dsc', \
 136             location = self.loc['contrib'], filesize = 0, md5sum = '')
 137         self.file['hello_2.2-1_i386.deb'] = PoolFile( \
 138             filename = 'main/h/hello/hello_2.2-1_i386.deb', \
 139             location = self.loc['main'], filesize = 0, md5sum = '')
 140         self.file['gnome-hello_2.2-1_i386.deb'] = PoolFile( \
 141             filename = 'main/h/hello/gnome-hello_2.2-1_i386.deb', \
 142             location = self.loc['main'], filesize = 0, md5sum = '')
 143         self.file['python-hello_2.2-1_all.deb'] = PoolFile( \
 144             filename = 'main/h/hello/python-hello_2.2-1_all.deb', \
 145             location = self.loc['main'], filesize = 0, md5sum = '')
 146         self.file['gnome-hello_3.0-1_i386.deb'] = PoolFile( \
 147             filename = 'main/g/gnome-hello/gnome-hello_3.0-1_i386.deb', \
 148             location = self.loc['contrib'], filesize = 0, md5sum = '')
 149         self.file['sl_3.03-16.dsc'] = PoolFile(filename = 'main/s/sl/sl_3.03-16.dsc', \
 150             location = self.loc['main'], filesize = 0, md5sum = '')
 151         self.file['python2.6_2.6.6-8.dsc'] = PoolFile( \
 152             filename = 'main/p/python2.6/python2.6_2.6.6-8.dsc', \
 153             location = self.loc['main'], filesize = 0, md5sum = '')
 154         self.session.add_all(self.file.values())
 155
 156     def setup_maintainers(self):
 157         'create some Maintainer objects'
 158
 159         if 'maintainer' in self.__dict__:
 160             return
 161         self.maintainer = {}
 162         self.maintainer['maintainer'] = Maintainer(name = 'Mr. Maintainer')
 163         self.maintainer['uploader'] = Maintainer(name = 'Mrs. Uploader')
 164         self.maintainer['lazyguy'] = Maintainer(name = 'Lazy Guy')
 165         self.session.add_all(self.maintainer.values())
 166
 167     def setup_sources(self):
 168         'create DBSource objects'
 169
 170         if 'source' in self.__dict__:
 171             return
 172         install_date = self.now()
 173         self.setup_maintainers()
 174         self.setup_suites()
 175         self.setup_poolfiles()
 176         self.source = {}
 177         self.source['hello_2.2-2'] = DBSource(source = 'hello', version = '2.2-2', \
 178             maintainer = self.maintainer['maintainer'], \
 179             changedby = self.maintainer['uploader'], \
 180             poolfile = self.file['hello_2.2-2.dsc'], install_date = install_date)
 181         self.source['hello_2.2-2'].suites.append(self.suite['sid'])
 182         self.source['hello_2.2-1'] = DBSource(source = 'hello', version = '2.2-1', \
 183             maintainer = self.maintainer['maintainer'], \
 184             changedby = self.maintainer['uploader'], \
 185             poolfile = self.file['hello_2.2-1.dsc'], install_date = install_date)
 186         self.source['hello_2.2-1'].suites.append(self.suite['sid'])
 187         self.source['gnome-hello_3.0-1'] = DBSource(source = 'gnome-hello', \
 188             version = '3.0-1', maintainer = self.maintainer['maintainer'], \
 189             changedby = self.maintainer['uploader'], \
 190             poolfile = self.file['gnome-hello_3.0-1.dsc'], install_date = install_date)
 191         self.source['gnome-hello_3.0-1'].suites.append(self.suite['sid'])
 192         self.source['sl_3.03-16'] = DBSource(source = 'sl', version = '3.03-16', \
 193             maintainer = self.maintainer['maintainer'], \
 194             changedby = self.maintainer['uploader'], \
 195             poolfile = self.file['sl_3.03-16.dsc'], install_date = install_date)
 196         self.source['sl_3.03-16'].suites.append(self.suite['squeeze'])
 197         self.source['sl_3.03-16'].suites.append(self.suite['sid'])
 198         self.session.add_all(self.source.values())
 199
 200     def setup_binaries(self):
 201         'create DBBinary objects'
 202
 203         if 'binary' in self.__dict__:
 204             return
 205         self.setup_sources()
 206         self.setup_architectures()
 207         self.binary = {}
 208         self.binary['hello_2.2-1_i386'] = DBBinary(package = 'hello', \
 209             source = self.source['hello_2.2-1'], version = '2.2-1', \
 210             maintainer = self.maintainer['maintainer'], \
 211             architecture = self.arch['i386'], \
 212             poolfile = self.file['hello_2.2-1_i386.deb'])
 213         self.binary['hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 214         self.binary['hello_2.2-1_i386'].suites.append(self.suite['sid'])
 215         self.binary['gnome-hello_2.2-1_i386'] = DBBinary(package = 'gnome-hello', \
 216             source = self.source['hello_2.2-1'], version = '2.2-1', \
 217             maintainer = self.maintainer['maintainer'], \
 218             architecture = self.arch['i386'], \
 219             poolfile = self.file['gnome-hello_2.2-1_i386.deb'])
 220         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 221         self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['sid'])
 222         self.binary['gnome-hello_3.0-1_i386'] = DBBinary(package = 'gnome-hello', \
 223             source = self.source['gnome-hello_3.0-1'], version = '3.0-1', \
 224             maintainer = self.maintainer['maintainer'], \
 225             architecture = self.arch['i386'], \
 226             poolfile = self.file['gnome-hello_3.0-1_i386.deb'])
 227         self.binary['gnome-hello_3.0-1_i386'].suites.append(self.suite['sid'])
 228         self.binary['python-hello_2.2-1_i386'] = DBBinary(package = 'python-hello', \
 229             source = self.source['hello_2.2-1'], version = '2.2-1', \
 230             maintainer = self.maintainer['maintainer'], \
 231             architecture = self.arch['all'], \
 232             poolfile = self.file['python-hello_2.2-1_all.deb'])
 233         self.binary['python-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
 234         self.session.add_all(self.binary.values())
 235
 236     def setup_overridetypes(self):
 237         '''
 238         Setup self.otype of class OverrideType.
 239         '''
 240         if 'otype' in self.__dict__:
 241             return
 242         self.otype = {}
 243         for type_ in ('deb', 'udeb'):
 244             self.otype[type_] = OverrideType(overridetype = type_)
 245         self.session.add_all(self.otype.values())
 246         self.session.flush()
 247
 248     def setup_sections(self):
 249         '''
 250         Setup self.section of class Section.
 251         '''
 252         if 'section' in self.__dict__:
 253             return
 254         self.section = {}
 255         self.section['python'] = Section(section = 'python')
 256         self.session.add_all(self.section.values())
 257         self.session.flush()
 258
 259     def setup_priorities(self):
 260         '''
 261         Setup self.prio of class Priority.
 262         '''
 263         if 'prio' in self.__dict__:
 264             return
 265         self.prio = {}
 266         self.prio['standard'] = Priority(priority = 'standard', level = 7)
 267         self.session.add_all(self.prio.values())
 268         self.session.flush()
 269
 270     def setup_overrides(self):
 271         '''
 272         Setup self.override of class Override.
 273         '''
 274         if 'override' in self.__dict__:
 275             return
 276         self.setup_suites()
 277         self.setup_components()
 278         self.setup_overridetypes()
 279         self.setup_sections()
 280         self.setup_priorities()
 281         self.override = {}
 282         self.override['hello_sid_main_udeb'] = Override(package = 'hello', \
 283             suite = self.suite['sid'], component = self.comp['main'], \
 284             overridetype = self.otype['udeb'], \
 285             section = self.section['python'], priority = self.prio['standard'])
 286         self.override['hello_squeeze_main_deb'] = Override(package = 'hello', \
 287             suite = self.suite['squeeze'], component = self.comp['main'], \
 288             overridetype = self.otype['deb'], \
 289             section = self.section['python'], priority = self.prio['standard'])
 290         self.override['hello_lenny_contrib_deb'] = Override(package = 'hello', \
 291             suite = self.suite['lenny'], component = self.comp['contrib'], \
 292             overridetype = self.otype['deb'], \
 293             section = self.section['python'], priority = self.prio['standard'])
 294         self.session.add_all(self.override.values())
 295         self.session.flush()
 296
 297     def setUp(self):
 298         if self.metadata is None:
 299             self.initialize()
 300         self.session = DBConn().session()
 301
 302     def now(self):
 303         """
 304         Returns the current time at the db server. Please note the function
 305         returns the same value as long as it is in the same transaction. You
 306         should self.session.rollback() (or commit) if you rely on getting a
 307         fresh timestamp.
 308         """
 309
 310         return self.session.query(func.now()).scalar()
 311
 312     def classes_to_clean(self):
 313         """
 314         The function classes_to_clean() returns a list of classes. All objects
 315         of each class will be deleted from the database in tearDown(). This
 316         function should be overridden in derived test cases as needed.
 317         """
 318         return ()
 319
 320     def tearDown(self):
 321         self.session.rollback()
 322         for class_ in self.classes_to_clean():
 323             for object_ in self.session.query(class_):
 324                 self.session.delete(object_)
 325         self.session.commit()
 326         # usually there is no need to drop all tables here
 327         #self.metadata.drop_all()
 328