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())
238 def setup_overridetypes(self):
240 Setup self.otype of class OverrideType.
242 if 'otype' in self.__dict__:
245 for type_ in ('deb', 'udeb'):
246 self.otype[type_] = OverrideType(overridetype = type_)
247 self.session.add_all(self.otype.values())
250 def setup_sections(self):
252 Setup self.section of class Section.
254 if 'section' in self.__dict__:
257 self.section['python'] = Section(section = 'python')
258 self.session.add_all(self.section.values())
261 def setup_priorities(self):
263 Setup self.prio of class Priority.
265 if 'prio' in self.__dict__:
268 self.prio['standard'] = Priority(priority = 'standard', level = 7)
269 self.session.add_all(self.prio.values())
272 def setup_overrides(self):
274 Setup self.override of class Override.
276 if 'override' in self.__dict__:
279 self.setup_components()
280 self.setup_overridetypes()
281 self.setup_sections()
282 self.setup_priorities()
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())
300 if self.metadata is None:
302 self.session = DBConn().session()
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
312 return self.session.query(func.now()).scalar()
314 def classes_to_clean(self):
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.
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()