+ Add a test.
Signed-off-by: Torsten Werner <twerner@debian.org>
from daklib.dbconn import *
from daklib import utils
from daklib.regexes import re_extract_src_version
+from daklib.cruft import *
################################################################################
################################################################################
def do_newer_version(lowersuite_name, highersuite_name, code, session):
- lowersuite = get_suite(lowersuite_name, session)
- if not lowersuite:
- return
-
- highersuite = get_suite(highersuite_name, session)
- if not highersuite:
- return
-
- # Check for packages in $highersuite obsoleted by versions in $lowersuite
- q = session.execute("""
-WITH highersuite_maxversion AS (SELECT s.source AS source, max(s.version) AS version
- FROM src_associations sa, source s
- WHERE sa.suite = :highersuite_id AND sa.source = s.id group by s.source)
-SELECT s.source, s.version AS lower, s2.version AS higher
- FROM src_associations sa, source s, source s2, src_associations sa2, highersuite_maxversion hm
- WHERE sa.suite = :highersuite_id AND sa2.suite = :lowersuite_id AND sa.source = s.id
- AND sa2.source = s2.id AND s.source = s2.source
- AND hm.source = s.source AND hm.version < s2.version
- AND s.version < s2.version""", {'lowersuite_id': lowersuite.suite_id,
- 'highersuite_id': highersuite.suite_id})
- ql = q.fetchall()
- if ql:
+ list = newer_version(lowersuite_name, highersuite_name, session)
+ if len(list) > 0:
nv_to_remove = []
- print "Newer version in %s" % lowersuite.suite_name
- print "-----------------" + "-" * len(lowersuite.suite_name)
+ title = "Newer version in %s" % lowersuite_name
+ print title
+ print "-" * len(title)
print
- for i in ql:
+ for i in list:
(source, higher_version, lower_version) = i
print " o %s (%s, %s)" % (source, higher_version, lower_version)
nv_to_remove.append(source)
print
print "Suggested command:"
- print " dak rm -m \"[auto-cruft] %s\" -s %s %s" % (code, highersuite.suite_name,
+ print " dak rm -m \"[auto-cruft] %s\" -s %s %s" % (code, highersuite_name,
" ".join(nv_to_remove))
print
--- /dev/null
+from daklib.dbconn import *
+
+from sqlalchemy import func
+
+def newer_version(lowersuite_name, highersuite_name, session):
+ '''
+ Finds newer versions in lowersuite_name than in highersuite_name. Returns a
+ list of tuples (source, higherversion, lowerversion) where higherversion is
+ the newest version from highersuite_name and lowerversion is the newest
+ version from lowersuite_name.
+ '''
+
+ lowersuite = get_suite(lowersuite_name, session)
+ highersuite = get_suite(highersuite_name, session)
+
+ query = session.query(DBSource.source, func.max(DBSource.version)). \
+ with_parent(highersuite).group_by(DBSource.source)
+
+ list = []
+ for (source, higherversion) in query:
+ lowerversion = session.query(func.max(DBSource.version)). \
+ filter_by(source = source).filter(DBSource.version > higherversion). \
+ with_parent(lowersuite).group_by(DBSource.source).scalar()
+ if lowerversion is not None:
+ list.append((source, higherversion, lowerversion))
+ return list
+
################################################################################
-# TODO: Why do we have a separate Location class? Can't it be fully integrated
-# into class Component?
class Location(ORMObject):
def __init__(self, path = None, component = None):
self.path = path
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, func, __version__
from sqlalchemy.exc import SADeprecationWarning
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()
--- /dev/null
+#!/usr/bin/env python
+
+from db_test import DBDakTestCase
+
+from daklib.dbconn import *
+from daklib.cruft import *
+
+import unittest
+
+class CruftTestCase(DBDakTestCase):
+ """
+ This class checks various functions of cruft-report.
+ """
+
+ def setUp(self):
+ super(CruftTestCase, self).setUp()
+ self.setup_binaries()
+ # flush to make sure that the setup is correct
+ self.session.flush()
+
+ def test_newer_version(self):
+ 'tests newer_version()'
+
+ list = newer_version('squeeze', 'sid', self.session)
+ self.assertEqual([], list)
+ self.file['sl_3.03-17.dsc'] = PoolFile(filename = 'main/s/sl/sl_3.03-17.dsc', \
+ location = self.loc['main'], filesize = 0, md5sum = '')
+ self.source['sl_3.03-17'] = DBSource(source = 'sl', version = '3.03-17', \
+ maintainer = self.maintainer['maintainer'], \
+ changedby = self.maintainer['uploader'], \
+ poolfile = self.file['sl_3.03-17.dsc'], install_date = self.now())
+ self.source['sl_3.03-17'].suites.append(self.suite['squeeze'])
+ list = newer_version('squeeze', 'sid', self.session)
+ self.assertEqual([('sl', '3.03-16', '3.03-17')], list)
+
+if __name__ == '__main__':
+ unittest.main()
from db_test import DBDakTestCase
-from daklib.dbconn import Architecture, Suite, get_suite_architectures, \
- get_architecture_suites, Maintainer, DBSource, Location, PoolFile, \
- check_poolfile, get_poolfile_like_name, get_source_in_suite, \
- get_suites_source_in, add_dsc_to_db, source_exists, DBBinary, \
- get_suites_binary_in, add_deb_to_db, Component, \
- get_component_by_package_suite
+from daklib.dbconn import *
from daklib.queue_install import package_to_suite
from daklib.queue import get_newest_source, get_suite_version_by_source, \
get_source_by_package_and_suite, get_suite_version_by_package
database.
"""
- 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):
super(PackageTestCase, self).setUp()
self.setup_binaries()
result = get_component_by_package_suite('foobar', ['sid'], \
session = self.session)
self.assertEqual(None, result)
- # test that the newest version is returend
+ # test that the newest version is returned
result = get_component_by_package_suite('gnome-hello', ['squeeze'], \
session = self.session)
self.assertEqual('main', result)