1 from base_test import DakTestCase, fixture
3 from daklib.config import Config
4 from daklib.dbconn import *
6 from sqlalchemy import create_engine, func, __version__
7 from sqlalchemy.exc import SADeprecationWarning
8 from sqlalchemy.schema import DDL
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']
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()"""
32 class DBDakTestCase(DakTestCase):
33 def execute(self, statement):
34 DDL(statement).execute(self.metadata.bind)
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))
46 if cnf["DB::Name"] in ('backports', 'obscurity', 'projectb'):
47 self.fail("You have configured an invalid database name: '%s'." % \
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"]
57 connstr = "postgres:///%s" % cnf["DB::Name"]
58 if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
59 connstr += "?port=%s" % cnf["DB::Port"]
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)
66 self.metadata.bind = create_engine(connstr)
67 self.metadata.create_all()
68 self.create_all_triggers()
70 def setup_suites(self):
71 "setup a hash of Suite objects in self.suite"
73 if 'suite' in self.__dict__:
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())
80 def setup_architectures(self):
81 "setup Architecture objects in self.arch and connect to suites"
83 if 'arch' in self.__dict__:
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()
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())
98 def setup_components(self):
99 'create some Component objects'
101 if 'comp' in self.__dict__:
104 self.comp['main'] = Component(component_name = 'main')
105 self.comp['contrib'] = Component(component_name = 'contrib')
106 self.session.add_all(self.comp.values())
108 def setup_locations(self):
109 'create some Location objects'
111 if 'loc' in self.__dict__:
113 self.setup_components()
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())
123 def setup_poolfiles(self):
124 'create some PoolFile objects'
126 if 'file' in self.__dict__:
128 self.setup_locations()
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())
158 def setup_maintainers(self):
159 'create some Maintainer objects'
161 if 'maintainer' in self.__dict__:
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())
169 def setup_sources(self):
170 'create DBSource objects'
172 if 'source' in self.__dict__:
174 install_date = self.now()
175 self.setup_maintainers()
177 self.setup_poolfiles()
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())
202 def setup_binaries(self):
203 'create DBBinary objects'
205 if 'binary' in self.__dict__:
208 self.setup_architectures()
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())
239 if self.metadata is None:
241 self.session = DBConn().session()
245 Returns the current time at the db server. Please note the function
246 returns the same value as long as it is in the same transaction. You
247 should self.session.rollback() (or commit) if you rely on getting a
251 return self.session.query(func.now()).scalar()
253 def classes_to_clean(self):
255 The function classes_to_clean() returns a list of classes. All objects
256 of each class will be deleted from the database in tearDown(). This
257 function should be overridden in derived test cases as needed.
262 self.session.rollback()
263 for class_ in self.classes_to_clean():
264 self.session.query(class_).delete()
265 self.session.commit()
266 # usually there is no need to drop all tables here
267 #self.metadata.drop_all()