tests/dbtest_packages.py

   1 #!/usr/bin/env python
   2
   3 from db_test import DBDakTestCase
   4 from base_test import fixture
   5
   6 from daklib.dbconn import *
   7 from daklib.queue_install import package_to_suite
   8 from daklib.queue import get_newest_source, get_suite_version_by_source, \
   9     get_source_by_package_and_suite, get_suite_version_by_package
  10
  11 from sqlalchemy.orm.exc import MultipleResultsFound
  12 import unittest
  13
  14 class Pkg():
  15     'fake package class used for testing'
  16
  17     def __init__(self):
  18         self.dsc = {}
  19         self.files = {}
  20         self.changes = {}
  21
  22 class Upload():
  23     'fake Upload class used for testing'
  24
  25     def __init__(self, pkg):
  26         self.pkg = pkg
  27
  28 class PackageTestCase(DBDakTestCase):
  29     """
  30     PackageTestCase checks the handling of source and binary packages in dak's
  31     database.
  32     """
  33
  34     def setUp(self):
  35         super(PackageTestCase, self).setUp()
  36         self.setup_binaries()
  37         # flush to make sure that the setup is correct
  38         self.session.flush()
  39
  40     def test_suite_architecture(self):
  41         # check the id for architectures source and all
  42         self.assertEqual(1, self.arch['source'].arch_id)
  43         self.assertEqual(2, self.arch['all'].arch_id)
  44         # check the many to many relation between Suite and Architecture
  45         self.assertEqual('source', self.suite['lenny'].architectures[0])
  46         self.assertEqual(4, len(self.suite['lenny'].architectures))
  47         self.assertEqual(3, len(self.arch['i386'].suites))
  48         # check the function get_suite_architectures()
  49         architectures = get_suite_architectures('lenny', session = self.session)
  50         self.assertEqual(4, len(architectures))
  51         self.assertTrue(self.arch['source'] in architectures)
  52         self.assertTrue(self.arch['all'] in architectures)
  53         self.assertTrue(self.arch['kfreebsd-i386'] not in architectures)
  54         architectures = get_suite_architectures('sid', session = self.session)
  55         self.assertEqual(5, len(architectures))
  56         self.assertTrue(self.arch['kfreebsd-i386'] in architectures)
  57         architectures = get_suite_architectures('lenny', skipsrc = True, session = self.session)
  58         self.assertEqual(3, len(architectures))
  59         self.assertTrue(self.arch['source'] not in architectures)
  60         architectures = get_suite_architectures('lenny', skipall = True, session = self.session)
  61         self.assertEqual(3, len(architectures))
  62         self.assertTrue(self.arch['all'] not in architectures)
  63         # check the function get_architecture_suites()
  64         suites = get_architecture_suites('i386', self.session)
  65         self.assertEqual(3, len(suites))
  66         self.assertTrue(self.suite['lenny'] in suites)
  67         suites = get_architecture_suites('kfreebsd-i386', self.session)
  68         self.assertEqual(2, len(suites))
  69         self.assertTrue(self.suite['lenny'] not in suites)
  70         # check overrides
  71         self.assertEqual(0, self.suite['lenny'].overrides.count())
  72
  73     def test_poolfiles(self):
  74         '''
  75         Test the relation of the classes PoolFile and Location.
  76
  77         The code needs some explaination. The property Location.files is not a
  78         list as in other relations because such a list would become rather
  79         huge. It is a query object that can be queried, filtered, and iterated
  80         as usual.  But list like methods like append() and remove() are
  81         supported as well which allows code like:
  82
  83         somelocation.files.append(somefile)
  84         '''
  85
  86         main = self.loc['main']
  87         contrib = self.loc['contrib']
  88         self.assertEqual(fixture('ftp/pool/'), main.path)
  89         count = len(self.file.keys()) - 2
  90         self.assertEqual(count, main.files.count())
  91         self.assertEqual(2, contrib.files.count())
  92         poolfile = main.files. \
  93                 filter(PoolFile.filename.like('%/hello/hello%')). \
  94                 order_by(PoolFile.filename)[0]
  95         self.assertEqual('main/h/hello/hello_2.2-1.dsc', poolfile.filename)
  96         self.assertEqual(main, poolfile.location)
  97         # test get()
  98         self.assertEqual(poolfile, \
  99                 self.session.query(PoolFile).get(poolfile.file_id))
 100         self.assertEqual(None, self.session.query(PoolFile).get(-1))
 101         # test remove() and append()
 102         main.files.remove(self.file['sl_3.03-16.dsc'])
 103         contrib.files.append(self.file['sl_3.03-16.dsc'])
 104         self.assertEqual(count - 1, main.files.count())
 105         self.assertEqual(3, contrib.files.count())
 106         # test fullpath
 107         self.assertEqual(fixture('ftp/pool/main/s/sl/sl_3.03-16.dsc'), \
 108             self.file['sl_3.03-16.dsc'].fullpath)
 109         # test check_poolfile()
 110         self.assertEqual((True, self.file['sl_3.03-16.dsc']), \
 111             check_poolfile('main/s/sl/sl_3.03-16.dsc', 0, '', \
 112                 contrib.location_id, self.session))
 113         # test string value of 2nd argument
 114         self.assertEqual((True, self.file['sl_3.03-16.dsc']), \
 115             check_poolfile('main/s/sl/sl_3.03-16.dsc', '0', '', \
 116                 contrib.location_id, self.session))
 117         self.assertEqual((False, None), \
 118             check_poolfile('foobar', 0, '', contrib.location_id, self.session))
 119         self.assertEqual((False, self.file['sl_3.03-16.dsc']), \
 120             check_poolfile('main/s/sl/sl_3.03-16.dsc', 42, '', \
 121                 contrib.location_id, self.session))
 122         self.assertEqual((False, self.file['sl_3.03-16.dsc']), \
 123             check_poolfile('main/s/sl/sl_3.03-16.dsc', 0, 'deadbeef', \
 124                 contrib.location_id, self.session))
 125         # test get_poolfile_like_name()
 126         self.assertEqual([self.file['sl_3.03-16.dsc']], \
 127             get_poolfile_like_name('sl_3.03-16.dsc', self.session))
 128         self.assertEqual([], get_poolfile_like_name('foobar', self.session))
 129
 130     def test_maintainers(self):
 131         '''
 132         tests relation between Maintainer and DBSource
 133
 134         TODO: add relations to changes_pending_source
 135         '''
 136
 137         maintainer = self.maintainer['maintainer']
 138         self.assertEqual(maintainer,
 139             self.session.query(Maintainer).get(maintainer.maintainer_id))
 140         uploader = self.maintainer['uploader']
 141         self.assertEqual(uploader,
 142             self.session.query(Maintainer).get(uploader.maintainer_id))
 143         lazyguy = self.maintainer['lazyguy']
 144         self.assertEqual(lazyguy,
 145             self.session.query(Maintainer).get(lazyguy.maintainer_id))
 146         self.assertEqual(4, len(maintainer.maintains_sources))
 147         self.assertTrue(self.source['hello_2.2-2'] in maintainer.maintains_sources)
 148         self.assertEqual(maintainer.changed_sources, [])
 149         self.assertEqual(uploader.maintains_sources, [])
 150         self.assertEqual(4, len(uploader.changed_sources))
 151         self.assertTrue(self.source['sl_3.03-16'] in uploader.changed_sources)
 152         self.assertEqual(lazyguy.maintains_sources, [])
 153         self.assertEqual(lazyguy.changed_sources, [])
 154
 155     def get_source_in_suite_fail(self):
 156         '''
 157         This function throws the MultipleResultsFound exception because
 158         get_source_in_suite is broken.
 159
 160         TODO: fix get_source_in_suite
 161         '''
 162
 163         return get_source_in_suite('hello', 'sid', self.session)
 164
 165     def test_sources(self):
 166         'test relation between DBSource and PoolFile or Suite'
 167
 168         # test PoolFile
 169         self.assertEqual(self.file['hello_2.2-2.dsc'], self.source['hello_2.2-2'].poolfile)
 170         self.assertEqual(self.source['hello_2.2-2'], self.file['hello_2.2-2.dsc'].source)
 171         self.assertEqual(None, self.file['python2.6_2.6.6-8.dsc'].source)
 172         # test Suite
 173         squeeze = self.session.query(Suite). \
 174             filter(Suite.sources.contains(self.source['sl_3.03-16'])). \
 175             order_by(Suite.suite_name)[1]
 176         self.assertEqual(self.suite['squeeze'], squeeze)
 177         self.assertEqual(1, squeeze.sources.count())
 178         self.assertEqual(self.source['sl_3.03-16'], squeeze.sources[0])
 179         sl = self.session.query(DBSource). \
 180             filter(DBSource.suites.contains(self.suite['squeeze'])).one()
 181         self.assertEqual(self.source['sl_3.03-16'], sl)
 182         self.assertEqual(2, len(sl.suites))
 183         self.assertTrue(self.suite['sid'] in sl.suites)
 184         # test get_source_in_suite()
 185         self.assertRaises(MultipleResultsFound, self.get_source_in_suite_fail)
 186         self.assertEqual(None, \
 187             get_source_in_suite('hello', 'squeeze', self.session))
 188         self.assertEqual(self.source['sl_3.03-16'], \
 189             get_source_in_suite('sl', 'sid', self.session))
 190         # test get_suites_source_in()
 191         self.assertEqual([self.suite['sid']], \
 192             get_suites_source_in('hello', self.session))
 193         self.assertEqual(2, len(get_suites_source_in('sl', self.session)))
 194         self.assertTrue(self.suite['squeeze'] in \
 195             get_suites_source_in('sl', self.session))
 196
 197     def test_add_dsc_to_db(self):
 198         'tests function add_dsc_to_db()'
 199
 200         pkg = Pkg()
 201         pkg.dsc['source'] = 'hello'
 202         pkg.dsc['version'] = '2.2-3'
 203         pkg.dsc['maintainer'] = self.maintainer['maintainer'].name
 204         pkg.changes['changed-by'] = self.maintainer['uploader'].name
 205         pkg.changes['fingerprint'] = 'deadbeef'
 206         pkg.changes['distribution'] = { 'sid': '' }
 207         pkg.files['hello_2.2-3.dsc'] = { \
 208             'component': 'main',
 209             'location id': self.loc['main'].location_id,
 210             'files id': self.file['hello_2.2-3.dsc'].file_id }
 211         pkg.dsc_files = {}
 212         upload = Upload(pkg)
 213         (source, dsc_component, dsc_location_id, pfs) = \
 214             add_dsc_to_db(upload, 'hello_2.2-3.dsc', self.session)
 215         self.assertEqual('hello', source.source)
 216         self.assertEqual('2.2-3', source.version)
 217         self.assertEqual('sid', source.suites[0].suite_name)
 218         self.assertEqual('main', dsc_component)
 219         self.assertEqual(self.loc['main'].location_id, dsc_location_id)
 220         self.assertEqual([], pfs)
 221
 222     def test_source_exists(self):
 223         'test function source_exists()'
 224
 225         hello = self.source['hello_2.2-2']
 226         self.assertTrue(source_exists(hello.source, hello.version, \
 227             suites = ['sid'], session = self.session))
 228         # binNMU
 229         self.assertTrue(source_exists(hello.source, hello.version + '+b7', \
 230             suites = ['sid'], session = self.session))
 231         self.assertTrue(not source_exists(hello.source, hello.version, \
 232             suites = ['lenny', 'squeeze'], session = self.session))
 233         self.assertTrue(not source_exists(hello.source, hello.version, \
 234             suites = ['lenny', 'sid'], session = self.session))
 235         self.assertTrue(not source_exists(hello.source, hello.version, \
 236             suites = ['sid', 'lenny'], session = self.session))
 237         self.assertTrue(not source_exists(hello.source, '0815', \
 238             suites = ['sid'], session = self.session))
 239         # 'any' suite
 240         self.assertTrue(source_exists(hello.source, hello.version, \
 241             session = self.session))
 242
 243     def test_package_to_suite(self):
 244         'test function package_to_suite()'
 245
 246         pkg = Pkg()
 247         pkg.changes = { 'distribution': {} }
 248         upload = Upload(pkg)
 249         self.assertTrue(not package_to_suite(upload, 'sid', self.session))
 250         pkg.changes['distribution'] = { 'sid': '' }
 251         pkg.changes['architecture'] = { 'source': '' }
 252         self.assertTrue(package_to_suite(upload, 'sid', self.session))
 253         pkg.changes['architecture'] = {}
 254         pkg.changes['source'] = self.source['hello_2.2-2'].source
 255         pkg.changes['version'] = self.source['hello_2.2-2'].version
 256         self.assertTrue(not package_to_suite(upload, 'sid', self.session))
 257         pkg.changes['version'] = '42'
 258         self.assertTrue(package_to_suite(upload, 'sid', self.session))
 259         pkg.changes['source'] = 'foobar'
 260         pkg.changes['version'] = self.source['hello_2.2-2'].version
 261         self.assertTrue(package_to_suite(upload, 'sid', self.session))
 262         pkg.changes['distribution'] = { 'lenny': '' }
 263         self.assertTrue(package_to_suite(upload, 'lenny', self.session))
 264
 265     def test_get_newest_source(self):
 266         'test function get_newest_source()'
 267
 268         import daklib.queue
 269         daklib.queue.dm_suites = ['sid']
 270         self.assertEqual(self.source['hello_2.2-2'], get_newest_source('hello', self.session))
 271         self.assertEqual(None, get_newest_source('foobar', self.session))
 272
 273     def test_get_suite_version_by_source(self):
 274         'test function get_suite_version_by_source()'
 275
 276         result = get_suite_version_by_source('hello', self.session)
 277         self.assertEqual(2, len(result))
 278         self.assertTrue(('sid', '2.2-1') in result)
 279         self.assertTrue(('sid', '2.2-2') in result)
 280         result = get_suite_version_by_source('sl', self.session)
 281         self.assertEqual(2, len(result))
 282         self.assertTrue(('squeeze', '3.03-16') in result)
 283         self.assertTrue(('sid', '3.03-16') in result)
 284
 285     def test_binaries(self):
 286         '''
 287         tests class DBBinary; TODO: test relation with Architecture, Maintainer,
 288         PoolFile, and Fingerprint
 289         '''
 290
 291         # test Suite relation
 292         self.assertEqual(3, self.suite['sid'].binaries.count())
 293         self.assertTrue(self.binary['hello_2.2-1_i386'] in \
 294             self.suite['sid'].binaries.all())
 295         self.assertEqual(0, self.suite['lenny'].binaries.count())
 296         # test DBSource relation
 297         self.assertEqual(3, len(self.source['hello_2.2-1'].binaries))
 298         self.assertTrue(self.binary['hello_2.2-1_i386'] in \
 299             self.source['hello_2.2-1'].binaries)
 300         self.assertEqual(0, len(self.source['hello_2.2-2'].binaries))
 301         # test get_suites_binary_in()
 302         self.assertEqual(2, len(get_suites_binary_in('hello', self.session)))
 303         self.assertTrue(self.suite['sid'] in \
 304             get_suites_binary_in('hello', self.session))
 305         self.assertEqual(2, len(get_suites_binary_in('gnome-hello', self.session)))
 306         self.assertTrue(self.suite['squeeze'] in \
 307             get_suites_binary_in('gnome-hello', self.session))
 308         self.assertEqual(0, len(get_suites_binary_in('sl', self.session)))
 309
 310     def test_add_deb_to_db(self):
 311         'tests function add_deb_to_db()'
 312
 313         pkg = Pkg()
 314         pkg.changes['fingerprint'] = 'deadbeef'
 315         pkg.changes['distribution'] = { 'sid': '' }
 316         pkg.files['hello_2.2-2_i386.deb'] = { \
 317             'package': 'hello',
 318             'version': '2.2-2',
 319             'maintainer': self.maintainer['maintainer'].name,
 320             'architecture': 'i386',
 321             'dbtype': 'deb',
 322             'pool name': 'main/h/hello/',
 323             'location id': self.loc['main'].location_id,
 324             'source package': 'hello',
 325             'source version': '2.2-2',
 326             'size': 0,
 327             'md5sum': 'deadbeef',
 328             'sha1sum': 'deadbeef',
 329             'sha256sum': 'deadbeef'}
 330         upload = Upload(pkg)
 331         bin, poolfile = add_deb_to_db(upload, 'hello_2.2-2_i386.deb', self.session)
 332         self.session.refresh(poolfile)
 333         self.session.refresh(poolfile.binary)
 334         self.assertEqual('main/h/hello/hello_2.2-2_i386.deb', poolfile.filename)
 335         self.assertEqual('hello', poolfile.binary.package)
 336         self.assertEqual('2.2-2', poolfile.binary.version)
 337         self.assertEqual(['sid'], poolfile.binary.suites)
 338         self.assertEqual('Mr. Maintainer', poolfile.binary.maintainer.name)
 339         self.assertEqual('i386', poolfile.binary.architecture.arch_string)
 340         self.assertEqual('deb', poolfile.binary.binarytype)
 341         self.assertEqual(self.loc['main'], poolfile.location)
 342         self.assertEqual(self.source['hello_2.2-2'], poolfile.binary.source)
 343         self.assertEqual(0, poolfile.filesize)
 344         self.assertEqual('deadbeef', poolfile.md5sum)
 345         self.assertEqual('deadbeef', poolfile.sha1sum)
 346         self.assertEqual('deadbeef', poolfile.sha256sum)
 347
 348     def test_get_source_by_package_and_suite(self):
 349         'test get_source_by_package_and_suite()'
 350
 351         query = get_source_by_package_and_suite('hello', 'sid', self.session)
 352         self.assertEqual(self.source['hello_2.2-1'], query.one())
 353         query = get_source_by_package_and_suite('gnome-hello', 'squeeze', self.session)
 354         self.assertEqual(self.source['hello_2.2-1'], query.one())
 355         query = get_source_by_package_and_suite('hello', 'hamm', self.session)
 356         self.assertEqual(0, query.count())
 357         query = get_source_by_package_and_suite('foobar', 'squeeze', self.session)
 358         self.assertEqual(0, query.count())
 359
 360     def test_get_suite_version_by_package(self):
 361         'test function get_suite_version_by_package()'
 362
 363         result = get_suite_version_by_package('hello', 'i386', self.session)
 364         self.assertEqual(2, len(result))
 365         self.assertTrue(('sid', '2.2-1') in result)
 366         result = get_suite_version_by_package('hello', 'amd64', self.session)
 367         self.assertEqual(0, len(result))
 368         result = get_suite_version_by_package('python-hello', 'i386', self.session)
 369         self.assertEqual([('squeeze', '2.2-1')], result)
 370         result = get_suite_version_by_package('python-hello', 'amd64', self.session)
 371         self.assertEqual([('squeeze', '2.2-1')], result)
 372
 373     def test_components(self):
 374         'test class Component'
 375
 376         self.assertEqual([self.loc['main']], self.comp['main'].location)
 377         self.assertEqual([self.loc['contrib']], self.comp['contrib'].location)
 378         self.assertEqual(0, self.comp['main'].overrides.count())
 379
 380     def test_get_component_by_package_suite(self):
 381         'test get_component_by_package_suite()'
 382
 383         result = get_component_by_package_suite('hello', ['sid'], \
 384             session = self.session)
 385         self.assertEqual('main', result)
 386         result = get_component_by_package_suite('hello', ['hamm'], \
 387             session = self.session)
 388         self.assertEqual(None, result)
 389         result = get_component_by_package_suite('foobar', ['sid'], \
 390             session = self.session)
 391         self.assertEqual(None, result)
 392         # test that the newest version is returned
 393         result = get_component_by_package_suite('gnome-hello', ['squeeze'], \
 394             session = self.session)
 395         self.assertEqual('main', result)
 396         result = get_component_by_package_suite('gnome-hello', ['sid'], \
 397             session = self.session)
 398         self.assertEqual('contrib', result)
 399         # test arch_list
 400         result = get_component_by_package_suite('hello', ['sid'], \
 401             arch_list = ['i386'], session = self.session)
 402         self.assertEqual('main', result)
 403         result = get_component_by_package_suite('hello', ['sid'], \
 404             arch_list = ['amd64'], session = self.session)
 405         self.assertEqual(None, result)
 406
 407 if __name__ == '__main__':
 408     unittest.main()