From: Torsten Werner Date: Wed, 19 Jan 2011 21:11:58 +0000 (+0100) Subject: Extend PackageTestCase. X-Git-Url: https://err.no/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=b330cb76fe5d7067655b34b755de70c6a26643d9;p=dak Extend PackageTestCase. Signed-off-by: Torsten Werner --- diff --git a/daklib/dbconn.py b/daklib/dbconn.py index ab97fdd2..e685d919 100755 --- a/daklib/dbconn.py +++ b/daklib/dbconn.py @@ -1084,13 +1084,16 @@ class PoolFile(object): def fullpath(self): return os.path.join(self.location.path, self.filename) + def is_valid(self, filesize = -1, md5sum = None):\ + return self.filesize == filesize and self.md5sum == md5sum + __all__.append('PoolFile') @session_wrapper def check_poolfile(filename, filesize, md5sum, location_id, session=None): """ Returns a tuple: - (ValidFileFound [boolean or None], PoolFile object or None) + (ValidFileFound [boolean], PoolFile object or None) @type filename: string @param filename: the filename of the file to check against the DB @@ -1106,34 +1109,24 @@ def check_poolfile(filename, filesize, md5sum, location_id, session=None): @rtype: tuple @return: Tuple of length 2. - - If more than one file found with that name: (C{None}, C{None}) - If valid pool file found: (C{True}, C{PoolFile object}) - If valid pool file not found: - (C{False}, C{None}) if no file found - (C{False}, C{PoolFile object}) if file found with size/md5sum mismatch """ - q = session.query(PoolFile).filter_by(filename=filename) - q = q.join(Location).filter_by(location_id=location_id) - - ret = None - - if q.count() > 1: - ret = (None, None) - elif q.count() < 1: - ret = (False, None) - else: - obj = q.one() - if obj.md5sum != md5sum or obj.filesize != int(filesize): - ret = (False, obj) - - if ret is None: - ret = (True, obj) + poolfile = session.query(Location).get(location_id). \ + files.filter_by(filename=filename).first() + valid = False + if poolfile and poolfile.is_valid(filesize = filesize, md5sum = md5sum): + valid = True - return ret + return (valid, poolfile) __all__.append('check_poolfile') +# TODO: the implementation can trivially be inlined at the place where the +# function is called @session_wrapper def get_poolfile_by_id(file_id, session=None): """ @@ -1146,12 +1139,7 @@ def get_poolfile_by_id(file_id, session=None): @return: either the PoolFile object or None """ - q = session.query(PoolFile).filter_by(file_id=file_id) - - try: - return q.one() - except NoResultFound: - return None + return session.query(PoolFile).get(file_id) __all__.append('get_poolfile_by_id') diff --git a/tests/dbtest_packages.py b/tests/dbtest_packages.py index 05ba4116..279d5a0f 100755 --- a/tests/dbtest_packages.py +++ b/tests/dbtest_packages.py @@ -3,7 +3,8 @@ from db_test import DBDakTestCase from daklib.dbconn import Architecture, Suite, get_suite_architectures, \ - get_architecture_suites, Maintainer, DBSource, Location, PoolFile + get_architecture_suites, Maintainer, DBSource, Location, PoolFile, \ + check_poolfile, get_poolfile_like_name import unittest @@ -129,6 +130,11 @@ class PackageTestCase(DBDakTestCase): filter(PoolFile.filename.like('%/hello/hello%')).one() self.assertEqual('main/h/hello/hello_2.2-2.dsc', poolfile.filename) self.assertEqual(location, poolfile.location) + # test get() + self.assertEqual(poolfile, \ + self.session.query(PoolFile).get(poolfile.file_id)) + self.assertEqual(None, self.session.query(PoolFile).get(-1)) + # test remove() and append() location.files.remove(self.file['sl']) # TODO: deletion should cascade automatically self.session.delete(self.file['sl']) @@ -143,6 +149,22 @@ class PackageTestCase(DBDakTestCase): # test fullpath self.assertEqual('/srv/ftp-master.debian.org/ftp/pool/main/s/sl/sl_3.03-16.dsc', \ self.file['sl'].fullpath) + # test check_poolfile() + self.assertEqual((True, self.file['sl']), \ + check_poolfile('main/s/sl/sl_3.03-16.dsc', 0, '', \ + location.location_id, self.session)) + self.assertEqual((False, None), \ + check_poolfile('foobar', 0, '', location.location_id, self.session)) + self.assertEqual((False, self.file['sl']), \ + check_poolfile('main/s/sl/sl_3.03-16.dsc', 42, '', \ + location.location_id, self.session)) + self.assertEqual((False, self.file['sl']), \ + check_poolfile('main/s/sl/sl_3.03-16.dsc', 0, 'deadbeef', \ + location.location_id, self.session)) + # test get_poolfile_like_name() + self.assertEqual([self.file['sl']], \ + get_poolfile_like_name('sl_3.03-16.dsc', self.session)) + self.assertEqual([], get_poolfile_like_name('foobar', self.session)) def setup_maintainers(self): 'create some Maintainer objects'