X-Git-Url: https://git.decadent.org.uk/gitweb/?a=blobdiff_plain;f=tests%2Fdb_test.py;h=f91c15a82b783a7963a1ad51023b00b7eb767be2;hb=504d14a6c25d43017c8a6a44f299819f2f1341e5;hp=d7f906cd34a147ad79916b283c0122be0761e34a;hpb=4830d9be143c7645ea932b53fae095e275ad7814;p=dak.git

diff --git a/tests/db_test.py b/tests/db_test.py
index d7f906cd..f91c15a8 100644
--- a/tests/db_test.py
+++ b/tests/db_test.py
@@ -1,9 +1,9 @@
 from base_test import DakTestCase, fixture
 
 from daklib.config import Config
-from daklib.dbconn import DBConn
+from daklib.dbconn import *
 
-from sqlalchemy import create_engine, __version__
+from sqlalchemy import create_engine, func, __version__
 from sqlalchemy.exc import SADeprecationWarning
 from sqlalchemy.schema import DDL
 
@@ -39,7 +39,9 @@ class DBDakTestCase(DakTestCase):
         for table in all_tables:
             self.execute(create_trigger % (table, table))
 
-    def setUp(self):
+    metadata = None
+
+    def initialize(self):
         cnf = Config()
         if cnf["DB::Name"] in ('backports', 'obscurity', 'projectb'):
             self.fail("You have configured an invalid database name: '%s'." % \
@@ -58,14 +60,194 @@ class DBDakTestCase(DakTestCase):
 
         pickle_filename = 'db-metadata-%s.pkl' % __version__
         pickle_file = open(fixture(pickle_filename), 'r')
-        self.metadata = pickle.load(pickle_file)
+        DBDakTestCase.metadata = pickle.load(pickle_file)
         self.metadata.ddl_listeners = pickle.load(pickle_file)
         pickle_file.close()
         self.metadata.bind = create_engine(connstr)
         self.metadata.create_all()
         self.create_all_triggers()
+
+    def setup_suites(self):
+        "setup a hash of Suite objects in self.suite"
+
+        if 'suite' in self.__dict__:
+            return
+        self.suite = {}
+        for suite_name in ('lenny', 'squeeze', 'sid'):
+            self.suite[suite_name] = Suite(suite_name = suite_name, version = '-')
+        self.session.add_all(self.suite.values())
+
+    def setup_architectures(self):
+        "setup Architecture objects in self.arch and connect to suites"
+
+        if 'arch' in self.__dict__:
+            return
+        self.setup_suites()
+        self.arch = {}
+        for arch_string in ('source', 'all', 'i386', 'amd64', 'kfreebsd-i386'):
+            self.arch[arch_string] = Architecture(arch_string)
+            if arch_string != 'kfreebsd-i386':
+                self.arch[arch_string].suites = self.suite.values()
+            else:
+                self.arch[arch_string].suites = [self.suite['squeeze'], self.suite['sid']]
+        # hard code ids for source and all
+        self.arch['source'].arch_id = 1
+        self.arch['all'].arch_id = 2
+        self.session.add_all(self.arch.values())
+
+    def setup_components(self):
+        'create some Component objects'
+
+        if 'comp' in self.__dict__:
+            return
+        self.comp = {}
+        self.comp['main'] = Component(component_name = 'main')
+        self.comp['contrib'] = Component(component_name = 'contrib')
+        self.session.add_all(self.comp.values())
+
+    def setup_locations(self):
+        'create some Location objects'
+
+        if 'loc' in self.__dict__:
+            return
+        self.setup_components()
+        self.loc = {}
+        self.loc['main'] = Location( \
+            path = '/srv/ftp-master.debian.org/ftp/pool/', \
+            component = self.comp['main'])
+        self.loc['contrib'] = Location( \
+            path = '/srv/ftp-master.debian.org/ftp/pool/', \
+            component = self.comp['contrib'])
+        self.session.add_all(self.loc.values())
+
+    def setup_poolfiles(self):
+        'create some PoolFile objects'
+
+        if 'file' in self.__dict__:
+            return
+        self.setup_locations()
+        self.file = {}
+        self.file['hello_2.2-3.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-3.dsc', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['hello_2.2-2.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-2.dsc', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['hello_2.2-1.dsc'] = PoolFile(filename = 'main/h/hello/hello_2.2-1.dsc', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['gnome-hello_3.0-1.dsc'] = PoolFile( \
+            filename = 'main/g/gnome-hello/gnome-hello_3.0-1.dsc', \
+            location = self.loc['contrib'], filesize = 0, md5sum = '')
+        self.file['hello_2.2-1_i386.deb'] = PoolFile( \
+            filename = 'main/h/hello/hello_2.2-1_i386.deb', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['gnome-hello_2.2-1_i386.deb'] = PoolFile( \
+            filename = 'main/h/hello/gnome-hello_2.2-1_i386.deb', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['python-hello_2.2-1_all.deb'] = PoolFile( \
+            filename = 'main/h/hello/python-hello_2.2-1_all.deb', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['gnome-hello_3.0-1_i386.deb'] = PoolFile( \
+            filename = 'main/g/gnome-hello/gnome-hello_3.0-1_i386.deb', \
+            location = self.loc['contrib'], filesize = 0, md5sum = '')
+        self.file['sl_3.03-16.dsc'] = PoolFile(filename = 'main/s/sl/sl_3.03-16.dsc', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.file['python2.6_2.6.6-8.dsc'] = PoolFile( \
+            filename = 'main/p/python2.6/python2.6_2.6.6-8.dsc', \
+            location = self.loc['main'], filesize = 0, md5sum = '')
+        self.session.add_all(self.file.values())
+
+    def setup_maintainers(self):
+        'create some Maintainer objects'
+
+        if 'maintainer' in self.__dict__:
+            return
+        self.maintainer = {}
+        self.maintainer['maintainer'] = Maintainer(name = 'Mr. Maintainer')
+        self.maintainer['uploader'] = Maintainer(name = 'Mrs. Uploader')
+        self.maintainer['lazyguy'] = Maintainer(name = 'Lazy Guy')
+        self.session.add_all(self.maintainer.values())
+
+    def setup_sources(self):
+        'create DBSource objects'
+
+        if 'source' in self.__dict__:
+            return
+        self.setup_maintainers()
+        self.setup_suites()
+        self.setup_poolfiles()
+        self.source = {}
+        self.source['hello_2.2-2'] = DBSource(source = 'hello', version = '2.2-2', \
+            maintainer = self.maintainer['maintainer'], \
+            changedby = self.maintainer['uploader'], \
+            poolfile = self.file['hello_2.2-2.dsc'], install_date = self.now())
+        self.source['hello_2.2-2'].suites.append(self.suite['sid'])
+        self.source['hello_2.2-1'] = DBSource(source = 'hello', version = '2.2-1', \
+            maintainer = self.maintainer['maintainer'], \
+            changedby = self.maintainer['uploader'], \
+            poolfile = self.file['hello_2.2-1.dsc'], install_date = self.now())
+        self.source['hello_2.2-1'].suites.append(self.suite['sid'])
+        self.source['gnome-hello_3.0-1'] = DBSource(source = 'gnome-hello', \
+            version = '3.0-1', maintainer = self.maintainer['maintainer'], \
+            changedby = self.maintainer['uploader'], \
+            poolfile = self.file['gnome-hello_3.0-1.dsc'], install_date = self.now())
+        self.source['gnome-hello_3.0-1'].suites.append(self.suite['sid'])
+        self.source['sl_3.03-16'] = DBSource(source = 'sl', version = '3.03-16', \
+            maintainer = self.maintainer['maintainer'], \
+            changedby = self.maintainer['uploader'], \
+            poolfile = self.file['sl_3.03-16.dsc'], install_date = self.now())
+        self.source['sl_3.03-16'].suites.append(self.suite['squeeze'])
+        self.source['sl_3.03-16'].suites.append(self.suite['sid'])
+        self.session.add_all(self.source.values())
+
+    def setup_binaries(self):
+        'create DBBinary objects'
+
+        if 'binary' in self.__dict__:
+            return
+        self.setup_sources()
+        self.setup_architectures()
+        self.binary = {}
+        self.binary['hello_2.2-1_i386'] = DBBinary(package = 'hello', \
+            source = self.source['hello_2.2-1'], version = '2.2-1', \
+            maintainer = self.maintainer['maintainer'], \
+            architecture = self.arch['i386'], \
+            poolfile = self.file['hello_2.2-1_i386.deb'])
+        self.binary['hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
+        self.binary['hello_2.2-1_i386'].suites.append(self.suite['sid'])
+        self.binary['gnome-hello_2.2-1_i386'] = DBBinary(package = 'gnome-hello', \
+            source = self.source['hello_2.2-1'], version = '2.2-1', \
+            maintainer = self.maintainer['maintainer'], \
+            architecture = self.arch['i386'], \
+            poolfile = self.file['gnome-hello_2.2-1_i386.deb'])
+        self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
+        self.binary['gnome-hello_2.2-1_i386'].suites.append(self.suite['sid'])
+        self.binary['gnome-hello_3.0-1_i386'] = DBBinary(package = 'gnome-hello', \
+            source = self.source['gnome-hello_3.0-1'], version = '3.0-1', \
+            maintainer = self.maintainer['maintainer'], \
+            architecture = self.arch['i386'], \
+            poolfile = self.file['gnome-hello_3.0-1_i386.deb'])
+        self.binary['gnome-hello_3.0-1_i386'].suites.append(self.suite['sid'])
+        self.binary['python-hello_2.2-1_i386'] = DBBinary(package = 'python-hello', \
+            source = self.source['hello_2.2-1'], version = '2.2-1', \
+            maintainer = self.maintainer['maintainer'], \
+            architecture = self.arch['all'], \
+            poolfile = self.file['python-hello_2.2-1_all.deb'])
+        self.binary['python-hello_2.2-1_i386'].suites.append(self.suite['squeeze'])
+        self.session.add_all(self.binary.values())
+
+    def setUp(self):
+        if self.metadata is None:
+            self.initialize()
         self.session = DBConn().session()
 
+    def now(self):
+        "returns the current time at the db server"
+
+        # we fetch a fresh session each time to avoid caching
+        local_session = DBConn().session()
+        current_time = local_session.query(func.now()).scalar()
+        local_session.close()
+        return current_time
+
     def classes_to_clean(self):
         """
         The function classes_to_clean() returns a list of classes. All objects
@@ -79,5 +261,6 @@ class DBDakTestCase(DakTestCase):
         for class_ in self.classes_to_clean():
             self.session.query(class_).delete()
         self.session.commit()
+        # usually there is no need to drop all tables here
         #self.metadata.drop_all()