launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #00480
[Merge] lp:~james-w/launchpad/more-matchers into lp:launchpad/devel
James Westby has proposed merging lp:~james-w/launchpad/more-matchers into lp:launchpad/devel with lp:~james-w/launchpad/improve-makeDistroArchSeries as a prerequisite.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
Hi,
This ports some Soyuz custom assertion methods to matchers.
I did this so that tests can make use of them without having to
subclass what is a fairly heavyweight test class.
I could have added a new TestCase subclass with the custom assertion
methods only and slotted it in to the existing hierarchy, but matchers
are better.
Lint:
./lib/lp/testing/factory.py
32: redefinition of unused 'os' from line 31
Thanks,
James
--
https://code.launchpad.net/~james-w/launchpad/more-matchers/+merge/32057
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~james-w/launchpad/more-matchers into lp:launchpad/devel.
=== modified file 'lib/lp/archivepublisher/tests/test_publisher.py'
--- lib/lp/archivepublisher/tests/test_publisher.py 2010-08-08 20:29:47 +0000
+++ lib/lp/archivepublisher/tests/test_publisher.py 2010-08-08 20:29:49 +0000
@@ -83,7 +83,6 @@
publisher.A_publish(False)
self.layer.txn.commit()
- removeSecurityProxy(pub_source).sync()
self.assertDirtyPocketsContents(
[('breezy-autotest', 'RELEASE')], publisher.dirty_pockets)
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
@@ -226,8 +225,6 @@
publisher.A_publish(force_publishing=False)
self.layer.txn.commit()
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_source2).sync()
self.assertDirtyPocketsContents(
[('hoary-test', 'RELEASE')], publisher.dirty_pockets)
self.assertEqual(pub_source2.status,
@@ -259,8 +256,6 @@
publisher.A_publish(force_publishing=False)
self.layer.txn.commit()
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_source2).sync()
self.assertDirtyPocketsContents(
[('breezy-autotest', 'UPDATES')], publisher.dirty_pockets)
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
@@ -360,7 +355,6 @@
publisher.A_publish(False)
self.layer.txn.commit()
- removeSecurityProxy(pub_source).sync()
self.assertDirtyPocketsContents(
[('breezy-autotest', 'RELEASE')], publisher.dirty_pockets)
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
=== modified file 'lib/lp/soyuz/doc/archive-files.txt'
--- lib/lp/soyuz/doc/archive-files.txt 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/doc/archive-files.txt 2010-08-08 20:29:49 +0000
@@ -143,6 +143,10 @@
>>> print another_test_source.sourcepackagerelease.package_diffs.count()
1
+We don't care about having a realistic diff for the two packages, for this
+test, and getting one through the normal means would be expensive, so
+we'll take a shortcut and set the content to be a dummy file.
+
>>> diff_name = 'test-pkg_1.0_1.1.diff.gz'
>>> diff = test_publisher.addMockFile(diff_name)
>>> removeSecurityProxy(package_diff).diff_content = diff
=== modified file 'lib/lp/soyuz/doc/packageupload-lookups.txt'
--- lib/lp/soyuz/doc/packageupload-lookups.txt 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/doc/packageupload-lookups.txt 2010-08-08 20:29:49 +0000
@@ -113,6 +113,7 @@
The `SourcePackageRelease` 'package_upload' and 'upload_changesfile'
>>> original_source_upload = source.sourcepackagerelease.package_upload
+ >>> # avoid the __repr__ of the security proxy
>>> print removeSecurityProxy(original_source_upload)
<PackageUpload ...>
=== modified file 'lib/lp/soyuz/doc/publishing.txt'
--- lib/lp/soyuz/doc/publishing.txt 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/doc/publishing.txt 2010-08-08 20:29:49 +0000
@@ -173,17 +173,36 @@
a rebuild archive then the status summary will always display
FULLY_BUILT.
+Create a COPY archive that will build for hppa and i386.
+
>>> from lp.soyuz.interfaces.archive import ArchivePurpose
- >>> removeSecurityProxy(spph.archive).purpose = ArchivePurpose.COPY
- >>> build_status_summary = spph.getStatusSummaryForBuilds()
+ >>> from lp.soyuz.interfaces.archivearch import IArchiveArchSet
+ >>> from lp.soyuz.interfaces.processor import IProcessorFamilySet
+ >>> archive = factory.makeArchive(
+ ... purpose=ArchivePurpose.COPY,
+ ... distribution=test_publisher.ubuntutest,
+ ... virtualized=False)
+ >>> processor_family = getUtility(IProcessorFamilySet).getByName('hppa')
+ >>> archive_arch_set = getUtility(IArchiveArchSet).new(
+ ... archive, processor_family)
+
+Publish a source to it, create the builds and mark them fully built.
+
+ >>> copy_spph = test_publisher.getPubSource(
+ ... sourcename="abc", version="666", archive=archive,
+ ... architecturehintlist='any')
+ >>> copy_builds = copy_spph.createMissingBuilds()
+ >>> for build in copy_builds:
+ ... build.status = BuildStatus.FULLYBUILT
+
+The check that the status summary marks all builds as FULLYBUILT.
+
+ >>> build_status_summary = copy_spph.getStatusSummaryForBuilds()
>>> print_build_status_summary(build_status_summary)
FULLYBUILT
hppa build of abc 666 in ubuntutest breezy-autotest RELEASE
i386 build of abc 666 in ubuntutest breezy-autotest RELEASE
- # Just set the purpose back before continuing on.
- >>> removeSecurityProxy(spph.archive).purpose = ArchivePurpose.PRIMARY
-
If one of the builds becomes published, it will not appear in the summary:
>>> from lp.soyuz.interfaces.publishing import (
@@ -780,6 +799,8 @@
This simulates a rebuild in of the same source in a more recent
distroseries, like rebuilding SRUs for constant sources.
+ >>> # removeSecurityProxy is required as no-one normally has the
+ >>> # permission to do this
>>> removeSecurityProxy(breezy_autotest).parent_series = None
>>> removeSecurityProxy(hoary_test).parent_series = breezy_autotest
@@ -792,6 +813,8 @@
Now, let's check the opposite, as if the copy was from a more recent
distroseries to a older one, like a backport rebuild.
+ >>> # removeSecurityProxy is required as no-one normally has the
+ >>> # permission to do this
>>> removeSecurityProxy(breezy_autotest).parent_series = hoary_test
>>> removeSecurityProxy(hoary_test).parent_series = None
@@ -1065,7 +1088,7 @@
main
>>> test_bin_pubs = test_publisher.getPubBinaries(
- ... pub_source=test_source_pub)
+ ... pub_source=test_source_pub, component='universe')
>>> test_bin_pub = test_bin_pubs[0]
>>> ppa_pub = publishing_set.newBinaryPublication(
... archive=mark.archive,
@@ -1810,7 +1833,10 @@
... sourcename='testing', version='2.0', component='universe',
... architecturehintlist='i386')
>>> test_binary = test_publisher.getPubBinaries(
- ... binaryname='testing-bin', pub_source=test_source)[0]
+ ... binaryname='testing-bin', pub_source=test_source,
+ ... component='multiverse')[0]
+ >>> test_binary = test_binary.changeOverride(
+ ... new_component=factory.makeComponent(name='universe'))
We will create a helper function to inspect ancestries. It simply pass
any given keyword argument to 'test_source' and 'test_binary'
@@ -1876,10 +1902,6 @@
It works in the same way for binaries.
- >>> multiverse = getUtility(IComponentSet)['multiverse']
- >>> naked_bpr = removeSecurityProxy(test_binary.binarypackagerelease)
- >>> naked_bpr.component = multiverse
-
>>> print test_binary.component.name
universe
=== modified file 'lib/lp/soyuz/doc/sourcepackagerelease.txt'
--- lib/lp/soyuz/doc/sourcepackagerelease.txt 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/doc/sourcepackagerelease.txt 2010-08-08 20:29:49 +0000
@@ -14,6 +14,7 @@
Let's get one from the database:
+ >>> from __future__ import with_statement
>>> from canonical.launchpad.database import SourcePackageRelease
>>> spr = SourcePackageRelease.get(14)
>>> spr.name
@@ -349,8 +350,9 @@
private PPA, it gets deleted from that private PPA. At this point the
package is still public:
- >>> from zope.security.proxy import removeSecurityProxy
- >>> removeSecurityProxy(private_publication).requestDeletion(cprov)
+ >>> from lp.testing import person_logged_in
+ >>> with person_logged_in(cprov):
+ ... private_publication.requestDeletion(cprov)
>>> transaction.commit()
>>> login('no-priv@xxxxxxxxxxxxx')
>>> check_permission('launchpad.View', test_sourcepackagerelease)
=== modified file 'lib/lp/soyuz/model/binarypackagename.py'
--- lib/lp/soyuz/model/binarypackagename.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/model/binarypackagename.py 2010-08-08 20:29:49 +0000
@@ -68,12 +68,6 @@
def new(self, name):
return BinaryPackageName(name=name)
- def getOrCreateByName(self, name):
- try:
- return self[name]
- except NotFoundError:
- return self.new(name)
-
def ensure(self, name):
"""Ensure that the given BinaryPackageName exists, creating it
if necessary.
@@ -81,9 +75,11 @@
Returns the BinaryPackageName
"""
try:
- return BinaryPackageName.byName(name)
- except SQLObjectNotFound:
- return BinaryPackageName(name=name)
+ return self[name]
+ except NotFoundError:
+ return self.new(name)
+
+ getOrCreateByName = ensure
def getNotNewByNames(self, name_ids, distroseries, archive_ids):
"""See `IBinaryPackageNameSet`."""
@@ -145,6 +141,8 @@
See sourcepackage.py:getSourcePackageDescriptions, which is analogous.
"""
+ if len(list(results)) < 1:
+ return {}
if use_names:
clause = ("BinaryPackageName.name in %s" %
sqlvalues([pn.name for pn in results]))
@@ -162,7 +160,7 @@
for release in releases:
binarypackagename = release.binarypackagename.name
- if binarypackagename in descriptions:
+ if binarypackagename not in descriptions:
description = release.description.strip().replace("\n", " ")
if len(description) > max_title_length:
description = (release.description[:max_title_length]
=== modified file 'lib/lp/soyuz/scripts/tests/test_changeoverride.py'
--- lib/lp/soyuz/scripts/tests/test_changeoverride.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/scripts/tests/test_changeoverride.py 2010-08-08 20:29:49 +0000
@@ -8,7 +8,6 @@
import unittest
from zope.component import getUtility
-from zope.security.proxy import removeSecurityProxy
from lp.soyuz.interfaces.component import IComponentSet
from lp.registry.interfaces.distribution import IDistributionSet
@@ -266,8 +265,7 @@
build = binaries[0].binarypackagerelease.build
other_binary = self.test_publisher.uploadBinaryForBuild(
- build, 'boingo-data')
- removeSecurityProxy(other_binary).version = '0.9'
+ build, 'boingo-data', version='0.9')
binaries.extend(
self.test_publisher.publishBinaryInArchive(
other_binary, source.archive, section="base",
=== modified file 'lib/lp/soyuz/scripts/tests/test_copypackage.py'
--- lib/lp/soyuz/scripts/tests/test_copypackage.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/scripts/tests/test_copypackage.py 2010-08-08 20:29:49 +0000
@@ -654,6 +654,24 @@
self.test_publisher = SoyuzTestPublisher()
self.test_publisher.prepareBreezyAutotest()
+ def makeTargetArchive(self):
+ return self.factory.makeArchive(
+ distribution=self.test_publisher.ubuntutest,
+ purpose=ArchivePurpose.PPA)
+
+ def test_checkCopy_can_copy_unexpired_binaries(self):
+ source = self.test_publisher.getPubSource()
+ binaries = self.test_publisher.getPubBinaries(pub_source=source)
+ archive = self.makeTargetArchive()
+ series = source.distroseries
+ pocket = source.pocket
+ copy_checker = CopyChecker(archive, include_binaries=False)
+ self.assertIs(
+ None, copy_checker.checkCopy(source, series, pocket))
+ copy_checker = CopyChecker(archive, include_binaries=True)
+ self.assertIs(
+ None, copy_checker.checkCopy(source, series, pocket))
+
def test_checkCopy_cannot_copy_expired_binaries(self):
# checkCopy() raises CannotCopy if the copy includes binaries
# and the binaries contain expired files. Publications of
@@ -661,30 +679,14 @@
# the file is unreachable.
# Create a testing source and binaries.
+ old_date = datetime.datetime(1970, 1, 1, tzinfo=pytz.UTC)
source = self.test_publisher.getPubSource()
- binaries = self.test_publisher.getPubBinaries(pub_source=source)
-
- # Create a fresh PPA which will be the destination copy.
- archive = self.factory.makeArchive(
- distribution=self.test_publisher.ubuntutest,
- purpose=ArchivePurpose.PPA)
+ binaries = self.test_publisher.getPubBinaries(
+ pub_source=source, files_expire=old_date)
+ archive = self.makeTargetArchive()
series = source.distroseries
pocket = source.pocket
- # At this point copy is allowed with or without binaries.
- copy_checker = CopyChecker(archive, include_binaries=False)
- self.assertIs(
- None, copy_checker.checkCopy(source, series, pocket))
- copy_checker = CopyChecker(archive, include_binaries=True)
- self.assertIs(
- None, copy_checker.checkCopy(source, series, pocket))
-
- # Set the expiration date of one of the testing binary files.
- utc = pytz.timezone('UTC')
- old_date = datetime.datetime(1970, 1, 1, tzinfo=utc)
- a_binary_file = binaries[0].binarypackagerelease.files[0]
- removeSecurityProxy(a_binary_file.libraryfile).expires = old_date
-
# Now source-only copies are allowed.
copy_checker = CopyChecker(archive, include_binaries=False)
self.assertIs(
@@ -702,20 +704,13 @@
# source that contain expired files. Publications of expired
# files can't be processed by the publisher since the file is
# unreachable.
- source = self.test_publisher.getPubSource()
-
- archive = self.factory.makeArchive(
- distribution=self.test_publisher.ubuntutest,
- purpose=ArchivePurpose.PPA)
+ expire = datetime.datetime.now(
+ pytz.UTC) + datetime.timedelta(days=365)
+ source = self.test_publisher.getPubSource(
+ files_expire=expire)
+ archive = self.makeTargetArchive()
series = source.distroseries
pocket = source.pocket
-
- utc = pytz.timezone('UTC')
- expire = datetime.datetime.now(utc) + datetime.timedelta(days=365)
-
- a_source_file = source.sourcepackagerelease.files[0]
- removeSecurityProxy(a_source_file.libraryfile).expires = expire
-
copy_checker = CopyChecker(archive, include_binaries=False)
self.assertRaisesWithContent(
CannotCopy,
@@ -974,7 +969,7 @@
self.test_publisher.breezy_autotest.status = (
SeriesStatus.CURRENT)
- def createDelayedCopyContext(self):
+ def createDelayedCopyContext(self, component='main'):
"""Create a context to allow delayed-copies test.
The returned source publication in a private archive with
@@ -986,7 +981,8 @@
ppa.buildd_secret = 'x'
ppa.private = True
- source = self.test_publisher.createSource(ppa, 'foocomm', '1.0-2')
+ source = self.test_publisher.createSource(
+ ppa, 'foocomm', '1.0-2', component=component)
self.test_publisher.getPubBinaries(pub_source=source)
[build] = source.getBuilds()
@@ -1060,14 +1056,13 @@
overridden when uploaded to the PPA, but when copying it to another
archive, only the ancestry in the destination archive can be used.
If that ancestry doesn't exist, an exception is raised."""
- # We'll simulate an upload that was overridden to main in the
- # ppa, by explicitly setting the spr's and bpr's component to
- # something else.
- source = self.createDelayedCopyContext()
- contrib = getUtility(IComponentSet).new('contrib')
- removeSecurityProxy(source.sourcepackagerelease).component = contrib
+ # Upload to contrib, with automatic override to main due
+ # to it being a PPA.
+ source = self.createDelayedCopyContext(component='contrib')
+ # But then override in to main, like a PPA would
[build] = source.getBuilds()
[binary] = build.binarypackages
+ contrib = self.factory.makeComponent(name="contrib")
binary.override(component=contrib)
self.layer.txn.commit()
@@ -1084,14 +1079,12 @@
archive, only the ancestry in the destination archive can be used.
If an ancestor is found in the destination archive, its component
is assumed for this package upload."""
- # We'll simulate an upload that was overridden to main in the
- # ppa, by explicitly setting the spr's and bpr's component to
- # something else.
- source = self.createDelayedCopyContext()
- contrib = getUtility(IComponentSet).new('contrib')
- removeSecurityProxy(source.sourcepackagerelease).component = contrib
+ # Upload to contrib, with automatic override to main due
+ # to it being a PPA.
+ source = self.createDelayedCopyContext(component='contrib')
[build] = source.getBuilds()
[binary] = build.binarypackages
+ contrib = self.factory.makeComponent(name='contrib')
binary.override(component=contrib)
# This time, we'll ensure that there is already an ancestor for
@@ -2234,7 +2227,16 @@
test_publisher.addFakeChroots(warty)
ppa_binaries = test_publisher.getPubBinaries(
pub_source=ppa_source, distroseries=warty,
- status=PackagePublishingStatus.PUBLISHED)
+ status=PackagePublishingStatus.PUBLISHED,
+ component='universe')
+
+ # Override the ppa binaries to main, like a PPA would
+ new_ppa_binaries = []
+ main = self.factory.makeComponent(name="main")
+ for binary in ppa_binaries:
+ new_ppa_binaries.append(
+ binary.changeOverride(new_component=main))
+ ppa_binaries = new_ppa_binaries
# Give the new source a private package diff.
sourcepackagerelease = other_source.sourcepackagerelease
@@ -2263,11 +2265,6 @@
universe = getUtility(IComponentSet)['universe']
ancestry_source.component = universe
- # Override the copied binarypackagerelease to 'universe'.
- for binary in ppa_binaries:
- removeSecurityProxy(binary.binarypackagerelease).component = (
- universe)
-
self.layer.txn.commit()
# Now we can invoke the unembargo script and check its results.
@@ -2365,10 +2362,9 @@
sourcename='buggy-source', version=version,
distroseries=warty, archive=archive, pocket=pocket,
changes_file_content=changes_file_content,
- status=PackagePublishingStatus.PUBLISHED)
- naked_spr = removeSecurityProxy(source.sourcepackagerelease)
- naked_spr.changelog_entry = (
- "Required for close_bugs_for_sourcepublication")
+ status=PackagePublishingStatus.PUBLISHED,
+ changelog_entry="Required for "
+ "close_bugs_for_sourcepublication")
binaries = test_publisher.getPubBinaries(
pub_source=source, distroseries=warty, archive=archive,
pocket=pocket, status=PackagePublishingStatus.PUBLISHED)
=== modified file 'lib/lp/soyuz/scripts/tests/test_publishdistro.py'
--- lib/lp/soyuz/scripts/tests/test_publishdistro.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/scripts/tests/test_publishdistro.py 2010-08-08 20:29:49 +0000
@@ -72,7 +72,6 @@
rc, out, err = self.runPublishDistroScript()
- removeSecurityProxy(pub_source).sync()
self.assertEqual(0, rc, "Publisher failed with:\n%s\n%s" % (out, err))
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
@@ -88,7 +87,6 @@
pub_source = self.getPubSource(filecontent='foo')
self.layer.txn.commit()
self.runPublishDistro()
- removeSecurityProxy(pub_source).sync()
random_person = getUtility(IPersonSet).getByName('name16')
pub_source.requestDeletion(random_person)
@@ -96,7 +94,6 @@
self.assertTrue(pub_source.scheduleddeletiondate is None,
"pub_source.scheduleddeletiondate should not be set, and it is.")
self.runPublishDistro()
- removeSecurityProxy(pub_source).sync()
self.assertTrue(pub_source.scheduleddeletiondate is not None,
"pub_source.scheduleddeletiondate should be set, and it's not.")
@@ -123,8 +120,6 @@
self.runPublishDistro(['-s', 'hoary-test'])
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_source2).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)
self.assertEqual(
pub_source2.status, PackagePublishingStatus.PUBLISHED)
@@ -212,9 +207,6 @@
self.runPublishDistro(['--ppa'])
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_source2).sync()
- removeSecurityProxy(pub_source3).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)
self.assertEqual(
pub_source2.status, PackagePublishingStatus.PUBLISHED)
@@ -253,14 +245,12 @@
# Try a plain PPA run, to ensure the private one is NOT published.
self.runPublishDistro(['--ppa'])
- removeSecurityProxy(pub_source).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)
# Now publish the private PPAs and make sure they are really
# published.
self.runPublishDistro(['--private-ppa'])
- removeSecurityProxy(pub_source).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
def testPublishPrimaryDebug(self):
@@ -288,6 +278,8 @@
# operation, see nascentupload-ddebs.txt.
self.prepareBreezyAutotest()
pub_binaries = self.getPubBinaries(format=BinaryPackageFormat.DDEB)
+ # changing the archive of a publication is not normally
+ # possible, so we have to use removeSecurityProxy.
for binary in pub_binaries:
removeSecurityProxy(binary).archive = debug_archive
=== added directory 'lib/lp/soyuz/testing'
=== added file 'lib/lp/soyuz/testing/__init__.py'
=== added file 'lib/lp/soyuz/testing/matchers.py'
--- lib/lp/soyuz/testing/matchers.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/testing/matchers.py 2010-08-08 20:29:49 +0000
@@ -0,0 +1,122 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+__all__ = [
+ 'IsSupersededBy',
+ 'PublishedStateIs',
+ 'PublishedStateIsNot',
+]
+
+from testtools.matchers import Annotate, Equals, Matcher, Mismatch
+
+from lp.soyuz.interfaces.publishing import PackagePublishingStatus
+from lp.testing.matchers import DateIsInPast
+
+
+class PublishedStateIsNot(Mismatch):
+
+ def __init__(self, publication, status):
+ """Create a PublishedStateIsNot Mismatch.
+
+ :param publication: the publication that has the wrong state.
+ :param status: the status that the publication should have had.
+ """
+ self.publication = publication
+ self.status = status
+
+ def describe(self):
+ return "Publication '%s' was '%s' instead of '%s'." % (
+ self.publication.displayname, self.publication.status.name,
+ self.status.name)
+
+
+class PublishedStateIs(Matcher):
+ """Check the state of publication(s) is a certain value."""
+
+ def __init__(self, status):
+ """Create a PublishedState matcher.
+
+ :param status: the status the publication(s) should have.
+ """
+ self.status = status
+
+ def __str__(self):
+ return "Published state is %s" % (self.status.name, )
+
+ def match(self, publications):
+ """Match the status of publication(s) against the expected.
+
+ :param publications: a publication or iterable of publications
+ of which to check the status.
+ :return: An instance of `PublishedStateIsNot` if any of the
+ publications have a different status, otherwise None if they
+ all match.
+ """
+ try:
+ list(publications)
+ except TypeError:
+ publications = [publications]
+ for publication in publications:
+ if publication.status != self.status:
+ return PublishedStateIsNot(publication, self.status)
+ return None
+
+
+class IsSupersededBy(Matcher):
+ """Check that superseded publishing record(s) have correct values.
+
+ A superseded publishing record should have:
+ status = PackagePublishingStatus.SUPERSEDED
+ datesuperseded in the past
+ superseded_by a particular publication or None
+ """
+
+ def __init__(self, superseded_by):
+ """Create an IsSupersededBy Matcher.
+
+ :param superseded_by: the publication that should have
+ supecseded the checked publications, or None if they
+ shouldn't have been superseded by another publication.
+ """
+ self.superseded_by = superseded_by
+
+ def __str__(self):
+ by = None
+ if self.superseded_by is not None:
+ by = "'%s'" % self.superseded_by.displayname
+ return "Is correctly superseded (by %s)." % by
+
+ def match(self, publications):
+ mismatch = PublishedStateIs(
+ PackagePublishingStatus.SUPERSEDED).match(publications)
+ if mismatch is not None:
+ return mismatch
+ try:
+ list(publications)
+ except TypeError:
+ publications = [publications]
+
+ def spr_title(spr):
+ if spr is None:
+ return None
+ else:
+ return spr.title
+ for publication in publications:
+ matcher = Annotate(
+ "'%s' has a datesuperseded in the future." % (
+ publication.displayname, ),
+ DateIsInPast())
+ mismatch = matcher.match(publication.datesuperseded)
+ if mismatch is not None:
+ return mismatch
+ matcher = Annotate(
+ "'%s' has the wrong supersededby, expected '%s', got '%s'"
+ % (publication.displayname, spr_title(self.superseded_by),
+ spr_title(publication.supersededby)),
+ Equals(self.superseded_by))
+ mismatch = matcher.match(publication.supersededby)
+ if mismatch is not None:
+ return mismatch
+ return None
=== added directory 'lib/lp/soyuz/testing/tests'
=== added file 'lib/lp/soyuz/testing/tests/__init__.py'
=== added file 'lib/lp/soyuz/testing/tests/test_matchers.py'
--- lib/lp/soyuz/testing/tests/test_matchers.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/testing/tests/test_matchers.py 2010-08-08 20:29:49 +0000
@@ -0,0 +1,244 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+from datetime import datetime, timedelta
+
+import pytz
+
+from testtools.matchers import AnnotatedMismatch
+
+from canonical.testing.layers import DatabaseFunctionalLayer
+from lp.soyuz.interfaces.publishing import PackagePublishingStatus
+from lp.soyuz.testing.matchers import (
+ IsSupersededBy, PublishedStateIs, PublishedStateIsNot)
+from lp.testing import celebrity_logged_in, TestCaseWithFactory
+from lp.testing.matchers import DateIsNotInPast
+
+
+class PublishedStateIsNotTests(TestCaseWithFactory):
+
+ layer = DatabaseFunctionalLayer
+
+ def test_describe_binary(self):
+ bpph = self.factory.makeBinaryPackagePublishingHistory(
+ status = PackagePublishingStatus.PENDING)
+ status = PackagePublishingStatus.PUBLISHED
+ mismatch = PublishedStateIsNot(bpph, status)
+ self.assertEqual(
+ "Publication '%s' was '%s' instead of '%s'." % (
+ bpph.displayname, bpph.status.name, status.name),
+ mismatch.describe())
+
+ def test_describe_source(self):
+ spph = self.factory.makeSourcePackagePublishingHistory(
+ status = PackagePublishingStatus.PENDING)
+ status = PackagePublishingStatus.PUBLISHED
+ mismatch = PublishedStateIsNot(spph, status)
+ self.assertEqual(
+ "Publication '%s' was '%s' instead of '%s'." % (
+ spph.displayname, spph.status.name, status.name),
+ mismatch.describe())
+
+
+class PublishedStateIsTests(TestCaseWithFactory):
+
+ layer = DatabaseFunctionalLayer
+
+ def test_str(self):
+ status = PackagePublishingStatus.PUBLISHED
+ matcher = PublishedStateIs(status)
+ self.assertEqual(
+ "Published state is %s" % (status.name, ),
+ str(matcher))
+
+ def getMatcherResultSingle(self, status):
+ spph = self.factory.makeSourcePackagePublishingHistory(
+ status=PackagePublishingStatus.PENDING)
+ matcher = PublishedStateIs(status)
+ return matcher.match(spph), spph
+
+ def getMatcherResultList(self, status):
+ pubs = [
+ self.factory.makeSourcePackagePublishingHistory(
+ status=status),
+ self.factory.makeSourcePackagePublishingHistory(
+ status=PackagePublishingStatus.PENDING),
+ ]
+ matcher = PublishedStateIs(status)
+ return matcher.match(pubs), pubs
+
+ def test_match_single(self):
+ mismatch, publication = self.getMatcherResultSingle(
+ PackagePublishingStatus.PENDING)
+ self.assertIs(None, mismatch)
+
+ def test_match_list(self):
+ mismatch, publications = self.getMatcherResultList(
+ PackagePublishingStatus.PENDING)
+ self.assertIs(None, mismatch)
+
+ def test_mismatch_single(self):
+ mismatch, publication = self.getMatcherResultSingle(
+ PackagePublishingStatus.PUBLISHED)
+ self.assertIsInstance(mismatch, PublishedStateIsNot)
+
+ def test_mismatch_list(self):
+ mismatch, publications = self.getMatcherResultList(
+ PackagePublishingStatus.PUBLISHED)
+ self.assertIsInstance(mismatch, PublishedStateIsNot)
+
+ def test_mismatch_sets_publication_single(self):
+ mismatch, publication = self.getMatcherResultSingle(
+ PackagePublishingStatus.PUBLISHED)
+ self.assertEqual(publication, mismatch.publication)
+
+ def test_mismatch_sets_publication_list(self):
+ mismatch, publications = self.getMatcherResultList(
+ PackagePublishingStatus.PUBLISHED)
+ self.assertIsInstance(mismatch, PublishedStateIsNot)
+ self.assertEqual(publications[1], mismatch.publication)
+
+ def test_mismatch_sets_status(self):
+ status = PackagePublishingStatus.PUBLISHED
+ mismatch, publication = self.getMatcherResultSingle(status)
+ self.assertEqual(status, mismatch.status)
+
+
+class IsSupersededByTests(TestCaseWithFactory):
+
+ layer = DatabaseFunctionalLayer
+
+ def test_str_with_superseded_by(self):
+ pub = self.factory.makeSourcePackagePublishingHistory()
+ matcher = IsSupersededBy(pub)
+ self.assertEqual(
+ "Is correctly superseded (by '%s')." % pub.displayname,
+ str(matcher))
+
+ def test_str_with_None(self):
+ matcher = IsSupersededBy(None)
+ self.assertEqual(
+ "Is correctly superseded (by None).", str(matcher))
+
+ def makeSupersededPublishing(self, superseded_by=None,
+ datesuperseded=None):
+ pub = self.factory.makeSourcePackagePublishingHistory(
+ status=PackagePublishingStatus.PUBLISHED)
+ pub.supersede(dominant=superseded_by)
+ if datesuperseded is not None:
+ with celebrity_logged_in('admin'):
+ pub.datesuperseded = datesuperseded
+ return pub
+
+ def test_match_single_superseded_by_not_None(self):
+ superseded_by = self.factory.makeSourcePackagePublishingHistory()
+ matcher = IsSupersededBy(superseded_by.sourcepackagerelease)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pub = self.makeSupersededPublishing(
+ superseded_by=superseded_by, datesuperseded=past)
+ mismatch = matcher.match(pub)
+ self.assertIs(None, mismatch)
+
+ def test_match_single_superseded_by_None(self):
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pub = self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past)
+ mismatch = matcher.match(pub)
+ self.assertIs(None, mismatch)
+
+ def test_match_list_superseded_by_not_None(self):
+ superseded_by = self.factory.makeSourcePackagePublishingHistory()
+ matcher = IsSupersededBy(superseded_by.sourcepackagerelease)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pubs = [
+ self.makeSupersededPublishing(
+ superseded_by=superseded_by, datesuperseded=past),
+ self.makeSupersededPublishing(
+ superseded_by=superseded_by, datesuperseded=past),
+ ]
+ mismatch = matcher.match(pubs)
+ self.assertIs(None, mismatch)
+
+ def test_match_list_superseded_by_None(self):
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pubs = [
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ ]
+ mismatch = matcher.match(pubs)
+ self.assertIs(None, mismatch)
+
+ def test_mismatch_single_wrong_status(self):
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pub = self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past)
+ with celebrity_logged_in('admin'):
+ pub.status = PackagePublishingStatus.PUBLISHED
+ mismatch = matcher.match(pub)
+ self.assertIsInstance(mismatch, PublishedStateIsNot)
+
+ def test_mismatch_list_wrong_status(self):
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pubs = [
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ ]
+ with celebrity_logged_in('admin'):
+ pubs[1].status = PackagePublishingStatus.PUBLISHED
+ mismatch = matcher.match(pubs)
+ self.assertIsInstance(mismatch, PublishedStateIsNot)
+
+ def test_mismatch_single_wrong_date(self):
+ matcher = IsSupersededBy(None)
+ future = datetime.now(pytz.UTC) + timedelta(days=4)
+ pub = self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=future)
+ mismatch = matcher.match(pub)
+ self.assertIsInstance(mismatch, AnnotatedMismatch)
+ self.assertIsInstance(mismatch.mismatch, DateIsNotInPast)
+
+ def test_mismatch_list_wrong_date(self):
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ future = datetime.now(pytz.UTC) + timedelta(days=4)
+ pubs = [
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=future),
+ ]
+ mismatch = matcher.match(pubs)
+ self.assertIsInstance(mismatch, AnnotatedMismatch)
+ self.assertIsInstance(mismatch.mismatch, DateIsNotInPast)
+
+ def test_mismatch_single_wrong_superseded_by(self):
+ superseded_by = self.factory.makeSourcePackagePublishingHistory()
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pub = self.makeSupersededPublishing(
+ superseded_by=superseded_by, datesuperseded=past)
+ mismatch = matcher.match(pub)
+ self.assertIsInstance(mismatch, AnnotatedMismatch)
+
+ def test_mismatch_list_wrong_superseded_by(self):
+ superseded_by = self.factory.makeSourcePackagePublishingHistory()
+ matcher = IsSupersededBy(None)
+ past = datetime.now(pytz.UTC) - timedelta(days=4)
+ pubs = [
+ self.makeSupersededPublishing(
+ superseded_by=None, datesuperseded=past),
+ self.makeSupersededPublishing(
+ superseded_by=superseded_by, datesuperseded=past),
+ ]
+ mismatch = matcher.match(pubs)
+ self.assertIsInstance(mismatch, AnnotatedMismatch)
=== added file 'lib/lp/soyuz/tests/test_binarypackagename.py'
--- lib/lp/soyuz/tests/test_binarypackagename.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/tests/test_binarypackagename.py 2010-08-08 20:29:49 +0000
@@ -0,0 +1,217 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test BinaryPackageName."""
+
+__metaclass__ = type
+
+from datetime import datetime
+
+import pytz
+from zope.component import getUtility
+
+from canonical.testing import DatabaseFunctionalLayer
+from lp.app.errors import NotFoundError
+from lp.soyuz.interfaces.binarypackagename import IBinaryPackageNameSet
+from lp.soyuz.interfaces.publishing import PackagePublishingStatus
+from lp.soyuz.model.binarypackagename import getBinaryPackageDescriptions
+from lp.testing import TestCaseWithFactory
+
+
+class TestBinaryPackageNameSet(TestCaseWithFactory):
+
+ layer = DatabaseFunctionalLayer
+
+ def setUp(self):
+ super(TestBinaryPackageNameSet, self).setUp()
+ self.name_set = getUtility(IBinaryPackageNameSet)
+
+ def test___getitem__found(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertEqual(name, self.name_set[name.name])
+
+ def test___getitem__not_found(self):
+ self.assertRaises(
+ NotFoundError, lambda name: self.name_set[name], "notfound")
+
+ def test_getAll_contains_one(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertIn(name, self.name_set.getAll())
+
+ def test_findByName_not_found(self):
+ self.assertEqual([], list(self.name_set.findByName("notfound")))
+
+ def test_findByName_found(self):
+ name1 = self.factory.makeBinaryPackageName("prefixname")
+ name2 = self.factory.makeBinaryPackageName("name")
+ name3 = self.factory.makeBinaryPackageName("namesuffix")
+ name4 = self.factory.makeBinaryPackageName("other")
+ self.assertEqual(
+ [name1, name2, name3], list(self.name_set.findByName("name")))
+
+ def test_queryByName_not_found(self):
+ self.assertEqual(None, self.name_set.queryByName("notfound"))
+
+ def test_queryByName_found(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertEqual(name, self.name_set.queryByName(name.name))
+
+ def test_new(self):
+ name = self.name_set.new("apackage")
+ self.assertEqual("apackage", name.name)
+
+ def test_getOrCreateByName_get(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertEqual(name, self.name_set.getOrCreateByName(name.name))
+
+ def test_getOrCreateByName_create(self):
+ self.assertEqual(
+ "apackage", self.name_set.getOrCreateByName("apackage").name)
+
+ def test_ensure_get(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertEqual(name, self.name_set.ensure(name.name))
+
+ def test_ensure_create(self):
+ self.assertEqual(
+ "apackage", self.name_set.ensure("apackage").name)
+
+ def createPublishingRecords(self, status=None):
+ distroseries = self.factory.makeDistroSeries()
+ distroarchseries = self.factory.makeDistroArchSeries(
+ distroseries=distroseries)
+ archives = [
+ self.factory.makeArchive(distribution=distroseries.distribution),
+ self.factory.makeArchive(distribution=distroseries.distribution),
+ ]
+ names = [
+ self.factory.makeBinaryPackageName(),
+ self.factory.makeBinaryPackageName(),
+ self.factory.makeBinaryPackageName(),
+ ]
+ for i in range(2):
+ bpr = self.factory.makeBinaryPackageRelease(
+ binarypackagename=names[i])
+ self.factory.makeBinaryPackagePublishingHistory(
+ binarypackagerelease=bpr,
+ status=PackagePublishingStatus.PUBLISHED,
+ archive=archives[i],
+ distroarchseries=distroarchseries,
+ )
+ return names, distroarchseries, archives
+
+ def test_getNotNewByNames_excludes_unpublished(self):
+ names, distroarchseries, archives = self.createPublishingRecords()
+ self.assertEqual(
+ sorted([names[0], names[1]]),
+ sorted(self.name_set.getNotNewByNames(
+ [name.id for name in names], distroarchseries.distroseries,
+ [archive.id for archive in archives])))
+
+ def test_getNotNewByNames_excludes_by_status(self):
+ names, distroarchseries, archives = self.createPublishingRecords()
+ bpr = self.factory.makeBinaryPackageRelease(
+ binarypackagename=names[2])
+ self.factory.makeBinaryPackagePublishingHistory(
+ binarypackagerelease=bpr,
+ status=PackagePublishingStatus.DELETED,
+ archive=archives[0], distroarchseries=distroarchseries)
+ self.assertEqual(
+ sorted([names[0], names[1]]),
+ sorted(self.name_set.getNotNewByNames(
+ [name.id for name in names], distroarchseries.distroseries,
+ [archive.id for archive in archives])))
+
+ def test_getNotNewByNames_excludes_by_name_id(self):
+ names, distroarchseries, archives = self.createPublishingRecords()
+ self.assertEqual(
+ [names[1]],
+ list(self.name_set.getNotNewByNames(
+ [name.id for name in names[1:]],
+ distroarchseries.distroseries,
+ [archive.id for archive in archives])))
+
+ def test_getNotNewByNames_excludes_by_distroseries(self):
+ names, distroarchseries, archives = self.createPublishingRecords()
+ bpr = self.factory.makeBinaryPackageRelease(
+ binarypackagename=names[2])
+ self.factory.makeBinaryPackagePublishingHistory(
+ binarypackagerelease=bpr,
+ status=PackagePublishingStatus.PUBLISHED,
+ archive=archives[0])
+ self.assertEqual(
+ sorted([names[0], names[1]]),
+ sorted(self.name_set.getNotNewByNames(
+ [name.id for name in names], distroarchseries.distroseries,
+ [archive.id for archive in archives])))
+
+ def test_getNotNewByNames_excludes_by_archive(self):
+ names, distroarchseries, archives = self.createPublishingRecords()
+ self.assertEqual(
+ [names[0]],
+ list(self.name_set.getNotNewByNames(
+ [name.id for name in names], distroarchseries.distroseries,
+ [archive.id for archive in archives[:1]])))
+
+ def test_getBinaryPackageDescriptions_none(self):
+ self.assertEqual({}, getBinaryPackageDescriptions([]))
+
+ def test_getBinaryPackageDescriptions_no_release(self):
+ name = self.factory.makeBinaryPackageName()
+ self.assertEqual({}, getBinaryPackageDescriptions([name]))
+
+ def test_getBinaryPackageDescriptions_one_release(self):
+ name = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="foo")
+ self.assertEqual(
+ {name.name: "foo"},
+ getBinaryPackageDescriptions([name], max_title_length=3))
+
+ def test_getBinaryPackageDescriptions_shortens_names(self):
+ name = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="foot")
+ self.assertEqual(
+ {name.name: "foo..."},
+ getBinaryPackageDescriptions([name], max_title_length=3))
+
+ def test_getBinaryPackageDescriptions_uses_latest(self):
+ name = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="foo",
+ date_created=datetime(1980, 01, 01, tzinfo=pytz.UTC))
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="bar",
+ date_created=datetime(2000, 01, 01, tzinfo=pytz.UTC))
+ self.assertEqual(
+ {name.name: "bar"},
+ getBinaryPackageDescriptions([name], max_title_length=3))
+
+ def test_getBinaryPackageDescriptions_two_packages(self):
+ name1 = self.factory.makeBinaryPackageName()
+ name2 = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name1, description="foo")
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name2, description="bar")
+ self.assertEqual(
+ {name1.name: "foo", name2.name: "bar"},
+ getBinaryPackageDescriptions([name1, name2], max_title_length=3))
+
+ def test_getBinaryPackageDescriptions_strips_newlines(self):
+ name = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="f\no")
+ self.assertEqual(
+ {name.name: "f o"},
+ getBinaryPackageDescriptions([name], max_title_length=3))
+
+ def test_getBinaryPackageDescriptions_use_names(self):
+ name = self.factory.makeBinaryPackageName()
+ self.factory.makeBinaryPackageRelease(
+ binarypackagename=name, description="foo")
+ self.assertEqual(
+ {name.name: "foo"},
+ getBinaryPackageDescriptions(
+ [name], use_names=True, max_title_length=3))
=== modified file 'lib/lp/soyuz/tests/test_publish_archive_indexes.py'
--- lib/lp/soyuz/tests/test_publish_archive_indexes.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/tests/test_publish_archive_indexes.py 2010-08-08 20:29:49 +0000
@@ -285,9 +285,8 @@
# An example of a corrupt dsc_binaries field. We need to ensure
# that the corruption is not carried over into the index stanza.
pub_source = self.getPubSource(
- sourcename="foo", version="666")
- naked_spr = removeSecurityProxy(pub_source.sourcepackagerelease)
- naked_spr.dsc_binaries = "foo_bin,\nbar_bin,\nzed_bin"
+ sourcename="foo", version="666",
+ dsc_binaries="foo_bin,\nbar_bin,\nzed_bin")
parser = self.write_stanza_and_reparse(pub_source.getIndexStanza())
=== modified file 'lib/lp/soyuz/tests/test_publishing.py'
--- lib/lp/soyuz/tests/test_publishing.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/tests/test_publishing.py 2010-08-08 20:29:49 +0000
@@ -29,8 +29,6 @@
from lp.registry.interfaces.sourcepackage import SourcePackageUrgency
from lp.registry.interfaces.sourcepackagename import ISourcePackageNameSet
from lp.soyuz.model.processor import ProcessorFamily
-from lp.soyuz.model.publishing import (
- BinaryPackagePublishingHistory)
from lp.soyuz.interfaces.archive import ArchivePurpose
from lp.soyuz.interfaces.archivearch import IArchiveArchSet
from lp.soyuz.interfaces.binarypackagerelease import BinaryPackageFormat
@@ -38,8 +36,10 @@
from lp.soyuz.interfaces.publishing import (
IPublishingSet, PackagePublishingPriority, PackagePublishingStatus)
from lp.soyuz.interfaces.queue import PackageUploadStatus
+from lp.soyuz.testing.matchers import IsSupersededBy, PublishedStateIs
from canonical.launchpad.scripts import FakeLogger
from lp.testing import TestCaseWithFactory
+from lp.testing.matcher import DateIsInPast
from lp.testing.factory import LaunchpadObjectFactory
from lp.testing.sampledata import UBUNTU_DEVELOPER_ADMIN_NAME
from lp.testing.fakemethod import FakeMethod
@@ -120,36 +120,21 @@
self.breezy_autotest_i386 = self.breezy_autotest['i386']
self.breezy_autotest_hppa = self.breezy_autotest['hppa']
- def addMockFile(self, filename, filecontent='nothing', restricted=False):
+ def addMockFile(self, filename, filecontent='nothing', restricted=False,
+ expires=None):
"""Add a mock file in Librarian.
Returns a ILibraryFileAlias corresponding to the file uploaded.
"""
return self.factory.makeLibraryFileAlias(
filename=filename, content=filecontent, restricted=restricted,
- content_type='application/text')
+ content_type='application/text', expires=expires)
def addPackageUpload(self, archive, distroseries,
pocket=PackagePublishingPocket.RELEASE,
changes_file_name="foo_666_source.changes",
changes_file_content="fake changes file content",
upload_status=PackageUploadStatus.DONE):
-<<<<<<< TREE
- signing_key = self.person.gpg_keys[0]
- package_upload = distroseries.createQueueEntry(
- pocket, changes_file_name, changes_file_content, archive,
- signing_key)
-
- status_to_method = {
- PackageUploadStatus.DONE: 'setDone',
- PackageUploadStatus.ACCEPTED: 'setAccepted',
- }
- naked_package_upload = removeSecurityProxy(package_upload)
- method = getattr(
- naked_package_upload, status_to_method[upload_status])
- method()
-
-=======
person = self.factory.makePerson()
signing_key = self.factory.makeGPGKey(person)
package_upload = self.factory.makePackageUpload(
@@ -157,7 +142,6 @@
changes_filename=changes_file_name,
changes_file_content=changes_file_content,
signing_key=signing_key, status=upload_status)
->>>>>>> MERGE-SOURCE
return package_upload
def getPubSource(self, sourcename=None, version='666', component='main',
@@ -175,7 +159,8 @@
build_conflicts_indep=None,
dsc_maintainer_rfc822='Foo Bar <foo@xxxxxxx>',
maintainer=None, creator=None, date_uploaded=UTC_NOW,
- spr_only=False):
+ spr_only=False, files_expire=None,
+ changelog_entry=None):
"""Return a mock source publishing record.
if spr_only is specified, the source is not published and the
@@ -224,33 +209,17 @@
dsc_standards_version=dsc_standards_version,
dsc_format=dsc_format,
dsc_binaries=dsc_binaries,
-<<<<<<< TREE
- archive=archive, dateuploaded=date_uploaded)
-
- changes_file_name = "%s_%s_source.changes" % (sourcename, version)
- if spr_only:
- upload_status = PackageUploadStatus.ACCEPTED
- else:
- upload_status = PackageUploadStatus.DONE
- package_upload = self.addPackageUpload(
- archive, distroseries, pocket,
- changes_file_name=changes_file_name,
- changes_file_content=changes_file_content,
- upload_status=upload_status)
- naked_package_upload = removeSecurityProxy(
- package_upload)
- naked_package_upload.addSource(spr)
-=======
archive=archive,
date_uploaded=date_uploaded,
+ changelog_entry=changelog_entry,
)
removeSecurityProxy(package_upload).addSource(spr)
->>>>>>> MERGE-SOURCE
if filename is None:
filename = "%s_%s.dsc" % (sourcename, version)
alias = self.addMockFile(
- filename, filecontent, restricted=archive.private)
+ filename, filecontent, restricted=archive.private,
+ expires=files_expire)
spr.addFile(alias)
if spr_only:
@@ -290,7 +259,7 @@
component='main',
section='base',
priority=PackagePublishingPriority.STANDARD,
- installed_size=100):
+ installed_size=100, files_expire=None):
"""Return a list of binary publishing records."""
if distroseries is None:
distroseries = self.distroseries
@@ -309,7 +278,7 @@
sourcename=sourcename, status=status, pocket=pocket,
archive=archive, distroseries=distroseries,
version=version, architecturehintlist=architecturehintlist,
- component=component)
+ component=component, files_expire=files_expire)
else:
archive = pub_source.archive
@@ -321,7 +290,8 @@
build, binaryname, filecontent, summary, description,
shlibdep, depends, recommends, suggests, conflicts, replaces,
provides, pre_depends, enhances, breaks, format,
- installed_size=installed_size)
+ installed_size=installed_size, component=component,
+ files_expire=files_expire)
pub_binaries = self.publishBinaryInArchive(
binarypackagerelease, archive, status, pocket,
scheduleddeletiondate, dateremoved, section, priority)
@@ -342,12 +312,15 @@
summary="summary", description="description", shlibdep=None,
depends=None, recommends=None, suggests=None, conflicts=None,
replaces=None, provides=None, pre_depends=None, enhances=None,
- breaks=None, format=BinaryPackageFormat.DEB, installed_size=None):
+ breaks=None, format=BinaryPackageFormat.DEB, installed_size=None,
+ component=None, version=None, files_expire=None):
"""Return the corresponding `BinaryPackageRelease`."""
sourcepackagerelease = build.source_package_release
distroarchseries = build.distro_arch_series
architecturespecific = (
not sourcepackagerelease.architecturehintlist == 'all')
+ if component is not None:
+ component = self.factory.makeComponent(name=component)
binarypackagename = self.factory.getOrMakeBinaryPackageName(
name=binaryname)
@@ -372,6 +345,8 @@
binpackageformat=format,
priority=PackagePublishingPriority.STANDARD,
installed_size=installed_size,
+ component=component,
+ version=version,
)
# Create the corresponding binary file.
@@ -383,7 +358,7 @@
filearchtag, format.name.lower())
alias = self.addMockFile(
filename, filecontent=filecontent,
- restricted=build.archive.private)
+ restricted=build.archive.private, expires=files_expire)
binarypackagerelease.addFile(alias)
# Adjust the build record in way it looks complete.
@@ -449,7 +424,7 @@
def createSource(
self, archive, sourcename, version, distroseries=None,
- new_version=None):
+ new_version=None, component='main'):
"""Create source with meaningful '.changes' file."""
top = 'lib/lp/archiveuploader/tests/data/suite'
name_fragment = '%s_%s' % (sourcename, version)
@@ -470,7 +445,7 @@
source = self.getPubSource(
sourcename=sourcename, archive=archive, version=new_version,
changes_file_content=changesfile_content,
- distroseries=distroseries)
+ distroseries=distroseries, component=component)
return source
@@ -524,17 +499,14 @@
def checkPublication(self, pub, status):
"""Assert the publication has the given status."""
- self.assertEqual(
- pub.status, status, "%s is not %s (%s)" % (
- pub.displayname, status.name, pub.status.name))
+ self.checkPublications(pub, status)
def checkPublications(self, pubs, status):
"""Assert the given publications have the given status.
See `checkPublication`.
"""
- for pub in pubs:
- self.checkPublication(pub, status)
+ self.assertThat(pubs, PublishedStateIs(status))
def checkPastDate(self, date, lag=None):
"""Assert given date is older than 'now'.
@@ -542,25 +514,10 @@
Optionally the user can pass a 'lag' which will be added to 'now'
before comparing.
"""
- UTC = pytz.timezone("UTC")
- limit = datetime.datetime.now(UTC)
- if lag is not None:
- limit = limit + lag
- self.assertTrue(date < limit, "%s >= %s" % (date, limit))
+ self.assertThat(date, DateIsInPast(lag=lag))
def checkSuperseded(self, pubs, supersededby=None):
- self.checkPublications(pubs, PackagePublishingStatus.SUPERSEDED)
- for pub in pubs:
- self.checkPastDate(pub.datesuperseded)
- if supersededby is not None:
- if isinstance(
- removeSecurityProxy(pub), BinaryPackagePublishingHistory):
- dominant = supersededby.binarypackagerelease.build
- else:
- dominant = supersededby.sourcepackagerelease
- self.assertEquals(dominant, pub.supersededby)
- else:
- self.assertIs(None, pub.supersededby)
+ self.assertThat(pubs, IsSupersededBy(supersededby))
class TestNativePublishing(TestNativePublishingBase):
@@ -625,7 +582,6 @@
self.layer.commit()
foo_name = "%s/main/f/foo/foo_666.dsc" % self.pool_dir
- removeSecurityProxy(pub_source).sync()
self.assertEqual(
pub_source.status, PackagePublishingStatus.PUBLISHED)
self.assertEqual(open(foo_name).read().strip(), 'foo is happy')
@@ -637,7 +593,6 @@
pub_source2.publish(self.disk_pool, self.logger)
self.layer.commit()
- removeSecurityProxy(pub_source2).sync()
self.assertEqual(
pub_source2.status, PackagePublishingStatus.PENDING)
self.assertEqual(open(foo_name).read().strip(), 'foo is happy')
@@ -654,7 +609,6 @@
self.layer.commit()
bar_name = "%s/main/b/bar/bar_666.dsc" % self.pool_dir
self.assertEqual(open(bar_name).read().strip(), 'bar is good')
- removeSecurityProxy(pub_source).sync()
self.assertEqual(
pub_source.status, PackagePublishingStatus.PUBLISHED)
@@ -662,7 +616,6 @@
sourcename='bar', filecontent='bar is good')
pub_source2.publish(self.disk_pool, self.logger)
self.layer.commit()
- removeSecurityProxy(pub_source2).sync()
self.assertEqual(
pub_source2.status, PackagePublishingStatus.PUBLISHED)
@@ -682,8 +635,6 @@
pub_source2.publish(self.disk_pool, self.logger)
self.layer.commit()
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_source2).sync()
self.assertEqual(
pub_source.status, PackagePublishingStatus.PUBLISHED)
self.assertEqual(
@@ -702,7 +653,6 @@
pub_source3.publish(self.disk_pool, self.logger)
self.layer.commit()
- removeSecurityProxy(pub_source3).sync()
self.assertEqual(
pub_source3.status, PackagePublishingStatus.PENDING)
@@ -723,7 +673,6 @@
pub_source.publish(test_disk_pool, self.logger)
self.layer.commit()
- removeSecurityProxy(pub_source).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
self.assertEqual(pub_source.sourcepackagerelease.upload_archive,
cprov.archive)
@@ -767,17 +716,21 @@
'Cannot override published records.',
source.overrideFromAncestry)
- def makeSource(self):
+ def makeSource(self, source_component='main', binary_component='main'):
"""Return a 'source' publication.
It's pending publication with binaries in a brand new PPA
- and in 'main' component.
+ and in 'main' component, unless otherwise specified. The binaries
+ can be placed in a different component to the source by
+ passing binary_component different to source_component.
"""
test_archive = self.factory.makeArchive(
distribution=self.test_publisher.ubuntutest,
purpose = ArchivePurpose.PPA)
- source = self.test_publisher.getPubSource(archive=test_archive)
- self.test_publisher.getPubBinaries(pub_source=source)
+ source = self.test_publisher.getPubSource(
+ archive=test_archive, component=source_component)
+ self.test_publisher.getPubBinaries(
+ pub_source=source, component=binary_component)
return source
def copyAndCheck(self, pub_record, series, component_name):
@@ -805,23 +758,14 @@
def test_overrideFromAncestry_fallback_to_source_component(self):
# overrideFromancestry on the lack of ancestry, falls back to the
# component the source was originally uploaded to.
- source = self.makeSource()
-
- # Adjust the source package release original component.
- universe = getUtility(IComponentSet)['universe']
- removeSecurityProxy(source.sourcepackagerelease).component = universe
-
+ source = self.makeSource(source_component='universe')
self.copyAndCheck(source, source.distroseries, 'universe')
def test_overrideFromAncestry_fallback_to_binary_component(self):
# overrideFromAncestry on the lack of ancestry, falls back to the
# component the binary was originally uploaded to.
- binary = self.makeSource().getPublishedBinaries()[0]
-
- # Adjust the binary package release original component.
- universe = getUtility(IComponentSet)['universe']
- removeSecurityProxy(binary.binarypackagerelease).component = universe
-
+ binary = self.makeSource(
+ binary_component='universe').getPublishedBinaries()[0]
self.copyAndCheck(
binary, binary.distroarchseries.distroseries, 'universe')
=== modified file 'lib/lp/soyuz/tests/test_publishing_top_level_api.py'
--- lib/lp/soyuz/tests/test_publishing_top_level_api.py 2010-08-08 20:29:47 +0000
+++ lib/lp/soyuz/tests/test_publishing_top_level_api.py 2010-08-08 20:29:49 +0000
@@ -3,8 +3,6 @@
"""Test top-level publication API in Soyuz."""
-from zope.security.proxy import removeSecurityProxy
-
from lp.soyuz.tests.test_publishing import TestNativePublishingBase
from lp.registry.interfaces.series import SeriesStatus
@@ -111,8 +109,6 @@
self._publish(pocket=pocket)
# source and binary PUBLISHED in database.
- removeSecurityProxy(pub_source).sync()
- removeSecurityProxy(pub_bin).sync()
self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)
self.assertEqual(pub_bin.status, PackagePublishingStatus.PUBLISHED)
=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py 2010-08-08 20:29:47 +0000
+++ lib/lp/testing/factory.py 2010-08-08 20:29:49 +0000
@@ -2243,7 +2243,7 @@
def makeSourcePackageName(self, name=None):
"""Make an `ISourcePackageName`."""
if name is None:
- name = self.getUniqueString()
+ name = self.getUniqueString('sourcepackagename')
return getUtility(ISourcePackageNameSet).new(name)
def getOrMakeSourcePackageName(self, name=None):
@@ -2289,14 +2289,13 @@
pocket, changes_filename, changes_file_content, archive,
signing_key=signing_key)
if status is not None:
- status_to_method = {
- PackageUploadStatus.DONE: 'setDone',
- PackageUploadStatus.ACCEPTED: 'setAccepted',
+ naked_package_upload = removeSecurityProxy(package_upload)
+ status_changers = {
+ PackageUploadStatus.DONE: naked_package_upload.setDone,
+ PackageUploadStatus.ACCEPTED:
+ naked_package_upload.setAccepted,
}
- naked_package_upload = removeSecurityProxy(package_upload)
- method = getattr(
- naked_package_upload, status_to_method[status])
- method()
+ status_changers[status]()
return package_upload
def makeSourcePackageRelease(self, archive=None, sourcepackagename=None,
@@ -2313,7 +2312,8 @@
dsc_format='1.0', dsc_binaries='foo-bin',
date_uploaded=UTC_NOW,
source_package_recipe_build=None,
- dscsigningkey=None):
+ dscsigningkey=None,
+ changelog_entry=None):
"""Make a `SourcePackageRelease`."""
if distroseries is None:
if source_package_recipe_build is not None:
@@ -2370,7 +2370,7 @@
build_conflicts_indep=build_conflicts_indep,
architecturehintlist=architecturehintlist,
changelog=None,
- changelog_entry=None,
+ changelog_entry=changelog_entry,
dsc=None,
copyright=self.getUniqueString(),
dscsigningkey=dscsigningkey,
@@ -2607,7 +2607,8 @@
conflicts=None, replaces=None,
provides=None, pre_depends=None,
enhances=None, breaks=None,
- essential=False, installed_size=None):
+ essential=False, installed_size=None,
+ date_created=None):
"""Make a `BinaryPackageRelease`."""
if build is None:
build = self.makeBinaryPackageBuild()
@@ -2628,7 +2629,7 @@
description = self.getUniqueString("description")
if installed_size is None:
installed_size = self.getUniqueInteger()
- return build.createBinaryPackageRelease(
+ bpr = build.createBinaryPackageRelease(
binarypackagename=binarypackagename, version=version,
binpackageformat=binpackageformat,
component=component, section=section, priority=priority,
@@ -2639,6 +2640,9 @@
provides=provides, pre_depends=pre_depends,
enhances=enhances, breaks=breaks, essential=essential,
installedsize=installed_size)
+ if date_created is not None:
+ removeSecurityProxy(bpr).datecreated = date_created
+ return bpr
def makeSection(self, name=None):
"""Make a `Section`."""
=== modified file 'lib/lp/testing/matchers.py'
--- lib/lp/testing/matchers.py 2010-08-08 20:29:47 +0000
+++ lib/lp/testing/matchers.py 2010-08-08 20:29:49 +0000
@@ -3,14 +3,12 @@
__metaclass__ = type
__all__ = [
+ 'DateIsInPast',
+ 'DateIsNotInPast',
'DoesNotCorrectlyProvide',
'DoesNotProvide',
-<<<<<<< TREE
- 'DoesNotCorrectlyProvide',
+ 'DoesNotStartWith',
'HasQueryCount',
-=======
- 'DoesNotStartWith',
->>>>>>> MERGE-SOURCE
'IsNotProxied',
'IsProxied',
'Provides',
@@ -18,6 +16,9 @@
'StartsWith',
]
+from datetime import datetime
+
+import pytz
from zope.interface.verify import verifyObject
from zope.interface.exceptions import (
BrokenImplementation, BrokenMethodImplementation, DoesNotImplement)
@@ -127,15 +128,15 @@
self.query_collector = query_collector
def describe(self):
- return "queries do not match: %s" % (self.count_mismatch.describe(),)
+ return "queries do not match: %s" % (self.count_mismatch.describe(), )
def get_details(self):
result = []
for query in self.query_collector.queries:
result.append(unicode(query).encode('utf8'))
return {'queries': Content(ContentType('text', 'plain',
- {'charset': 'utf8'}), lambda:['\n'.join(result)])}
-
+ {'charset': 'utf8'}), lambda: ['\n'.join(result)])}
+
class IsNotProxied(Mismatch):
"""An object is not proxied."""
@@ -217,3 +218,56 @@
if not matchee.startswith(self.expected):
return DoesNotStartWith(matchee, self.expected)
return None
+
+
+class DateIsNotInPast(Mismatch):
+
+ def __init__(self, date, limit):
+ """Create a DateIsNotInPast Mismatch.
+
+ :param date: the date that was checked.
+ :param limit: the date that it should have been before.
+ """
+ self.date = date
+ self.limit = limit
+
+ def describe(self):
+ return "%s >= %s" % (self.date, self.limit)
+
+
+class DateIsInPast(Matcher):
+ """A matcher that checks a datetime is in the past."""
+
+ def __init__(self, current_date=None, lag=None):
+ """Create a DateIsInPast Matcher.
+
+ :param current_date: the `datetime.datetime` that the date should be
+ before, or None for the current time when the check is done.
+ :param lag: a `datetime.timedelta` to add to the current time
+ before comparing, or None for no change.
+ """
+ self.current_date = current_date
+ self.lag = lag
+
+ def __str__(self):
+ if self.current_date is None:
+ start_str = 'UTC_NOW'
+ if self.lag is not None:
+ start_str += ' plus %s' % self.lag
+ else:
+ start = self.current_date
+ if self.lag is not None:
+ start += self.lag
+ start_str = str(start)
+ return "Date is before %s." % start_str
+
+ def match(self, date):
+ if self.current_date is None:
+ limit = datetime.now(pytz.UTC)
+ else:
+ limit = self.current_date
+ if self.lag is not None:
+ limit += self.lag
+ if date >= limit:
+ return DateIsNotInPast(date, limit)
+ return None
=== modified file 'lib/lp/testing/tests/test_factory.py'
--- lib/lp/testing/tests/test_factory.py 2010-08-08 20:29:47 +0000
+++ lib/lp/testing/tests/test_factory.py 2010-08-08 20:29:49 +0000
@@ -20,6 +20,7 @@
from lp.registry.interfaces.distribution import IDistribution
from lp.registry.interfaces.distroseries import IDistroSeries
from lp.registry.interfaces.sourcepackage import SourcePackageFileType
+from lp.registry.interfaces.sourcepackagename import ISourcePackageName
from lp.registry.interfaces.suitesourcepackage import ISuiteSourcePackage
from lp.services.worlddata.interfaces.language import ILanguage
from lp.soyuz.interfaces.binarypackagebuild import IBinaryPackageBuild
@@ -284,6 +285,12 @@
bpr = self.factory.makeBinaryPackageRelease(installed_size=110)
self.assertEqual(110, bpr.installedsize)
+ def test_makeBinaryPackageName_uses_date_created(self):
+ date_created = datetime(2000, 01, 01, tzinfo=pytz.UTC)
+ bpr = self.factory.makeBinaryPackageRelease(
+ date_created=date_created)
+ self.assertEqual(date_created, bpr.datecreated)
+
# makeCodeImport
def test_makeCodeImportNoStatus(self):
# If makeCodeImport is not given a review status, it defaults to NEW.
@@ -395,6 +402,15 @@
# And name is constructed from code as 'Language %(code)s'.
self.assertEquals('Test language', language.englishname)
+ # makeSourcePackageName
+ def test_makeSourcePackageName_returns_proxied_ISPN(self):
+ spn = self.factory.makeSourcePackageName()
+ self.assertThat(spn, ProvidesAndIsProxied(ISourcePackageName))
+
+ def test_makeSourcePackageName_created_has_useful_prefix(self):
+ spn = self.factory.makeSourcePackageName()
+ self.assertThat(spn.name, StartsWith("sourcepackagename"))
+
# makeSourcePackagePublishingHistory
def test_makeSourcePackagePublishingHistory_returns_ISPPH(self):
spph = self.factory.makeSourcePackagePublishingHistory()
@@ -459,6 +475,17 @@
dsc_maintainer_rfc822=maintainer)
self.assertEqual(maintainer, spr.dsc_maintainer_rfc822)
+ def test_makeSourcePackageRelease_allows_None_changelog_entry(self):
+ spr = self.factory.makeSourcePackageRelease(
+ changelog_entry=None)
+ self.assertEqual(None, spr.changelog_entry)
+
+ def test_makeSourcePackageRelease_uses_changelog_entry(self):
+ changelog_entry = "Best release evar!!1!1!"
+ spr = self.factory.makeSourcePackageRelease(
+ changelog_entry=changelog_entry)
+ self.assertEqual(changelog_entry, spr.changelog_entry)
+
# makeSuiteSourcePackage
def test_makeSuiteSourcePackage_returns_ISuiteSourcePackage(self):
ssp = self.factory.makeSuiteSourcePackage()
=== modified file 'lib/lp/testing/tests/test_matchers.py'
--- lib/lp/testing/tests/test_matchers.py 2010-08-08 20:29:47 +0000
+++ lib/lp/testing/tests/test_matchers.py 2010-08-08 20:29:49 +0000
@@ -3,6 +3,9 @@
__metaclass__ = type
+from datetime import datetime, timedelta
+
+import pytz
from zope.interface import implements, Interface
from zope.interface.verify import verifyObject
from zope.interface.exceptions import BrokenImplementation
@@ -11,16 +14,12 @@
from lp.testing import TestCase
from lp.testing.matchers import (
-<<<<<<< TREE
- DoesNotCorrectlyProvide, DoesNotProvide, HasQueryCount, IsNotProxied,
- IsProxied, Provides, ProvidesAndIsProxied)
+ DateIsInPast, DateIsNotInPast, DoesNotCorrectlyProvide, DoesNotProvide,
+ DoesNotStartWith, HasQueryCount, IsNotProxied, IsProxied, Provides,
+ ProvidesAndIsProxied, StartsWith)
from lp.testing._webservice import QueryCollector
from testtools.matchers import Is, Not, LessThan
-=======
- DoesNotCorrectlyProvide, DoesNotProvide, DoesNotStartWith, IsNotProxied,
- IsProxied, Provides, ProvidesAndIsProxied, StartsWith)
->>>>>>> MERGE-SOURCE
class ITestInterface(Interface):
@@ -179,7 +178,6 @@
obj = ProxyFactory(object(), checker=NamesChecker())
matcher = ProvidesAndIsProxied(ITestInterface)
self.assertIsInstance(matcher.match(obj), DoesNotProvide)
-<<<<<<< TREE
class TestQueryMatching(TestCase):
@@ -193,7 +191,7 @@
def test_match(self):
matcher = HasQueryCount(Is(3))
collector = QueryCollector()
- collector.count = 3
+ collector.count = 3
# not inspected
del collector.queries
self.assertThat(matcher.match(collector), Is(None))
@@ -214,13 +212,10 @@
self.assertEqual(["('foo', 'bar')\n('baaz', 'quux')"],
lines)
self.assertEqual(
- "queries do not match: %s" % (LessThan(2).match(2).describe(),),
+ "queries do not match: %s" % (LessThan(2).match(2).describe(), ),
mismatch.describe())
-=======
-
-
class DoesNotStartWithTests(TestCase):
def test_describe(self):
@@ -252,4 +247,95 @@
matcher = StartsWith("bar")
mismatch = matcher.match("foo")
self.assertEqual("bar", mismatch.expected)
->>>>>>> MERGE-SOURCE
+
+
+class DateIsNotInPastTests(TestCase):
+
+ def test_describe(self):
+ date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ limit = datetime(2000, 02, 02, 02, 02, 02, 02, pytz.UTC)
+ mismatch = DateIsNotInPast(date, limit)
+ self.assertEqual("%s >= %s" % (date, limit), mismatch.describe())
+
+
+class DateIsInPastTests(TestCase):
+
+ def test_str_now_no_lag(self):
+ matcher = DateIsInPast()
+ self.assertEqual("Date is before UTC_NOW.", str(matcher))
+
+ def test_str_now_with_lag(self):
+ lag = timedelta(seconds=60)
+ matcher = DateIsInPast(lag=lag)
+ self.assertEqual(
+ "Date is before UTC_NOW plus %s." % lag, str(matcher))
+
+ def test_str_current_date_no_lag(self):
+ date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ matcher = DateIsInPast(current_date=date)
+ self.assertEqual("Date is before %s." % date, str(matcher))
+
+ def test_str_current_date_with_lag(self):
+ date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ lag = timedelta(seconds=60)
+ matcher = DateIsInPast(current_date=date, lag=lag)
+ self.assertEqual("Date is before %s." % (date + lag, ), str(matcher))
+
+ def test_match_now_no_lag(self):
+ date = datetime.now(pytz.UTC) - timedelta(days=3)
+ matcher = DateIsInPast()
+ self.assertEqual(None, matcher.match(date))
+
+ def test_match_now_with_lag(self):
+ date = datetime.now(pytz.UTC) + timedelta(days=3)
+ lag = timedelta(days=4)
+ matcher = DateIsInPast(lag=lag)
+ self.assertEqual(None, matcher.match(date))
+
+ def test_current_date_no_lag(self):
+ current_date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ date = current_date - timedelta(days=3)
+ matcher = DateIsInPast(current_date=current_date)
+ self.assertEqual(None, matcher.match(date))
+
+ def test_current_date_with_lag(self):
+ current_date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ date = current_date + timedelta(days=3)
+ lag = timedelta(days=4)
+ matcher = DateIsInPast(current_date=current_date, lag=lag)
+ self.assertEqual(None, matcher.match(date))
+
+ def test_mismatch_now_no_lag(self):
+ date = datetime.now(pytz.UTC) + timedelta(days=3)
+ matcher = DateIsInPast()
+ self.assertIsInstance(matcher.match(date), DateIsNotInPast)
+
+ def test_mismatch_now_with_lag(self):
+ date = datetime.now(pytz.UTC) + timedelta(days=3)
+ lag = timedelta(days=2)
+ matcher = DateIsInPast(lag=lag)
+ self.assertIsInstance(matcher.match(date), DateIsNotInPast)
+
+ def test_mismatch_current_date_no_lag(self):
+ current_date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ date = current_date + timedelta(days=3)
+ matcher = DateIsInPast(current_date=current_date)
+ self.assertIsInstance(matcher.match(date), DateIsNotInPast)
+
+ def test_mismatch_current_date_with_lag(self):
+ current_date = datetime(2008, 01, 01, 01, 01, 01, 01, pytz.UTC)
+ date = current_date + timedelta(days=3)
+ lag = timedelta(days=2)
+ matcher = DateIsInPast(current_date=current_date, lag=lag)
+ self.assertIsInstance(matcher.match(date), DateIsNotInPast)
+
+ def test_mismatch_sets_date(self):
+ date = datetime.now(pytz.UTC) + timedelta(days=3)
+ matcher = DateIsInPast()
+ self.assertEqual(date, matcher.match(date).date)
+
+ def test_mismatch_sets_limit(self):
+ current_date = datetime.now(pytz.UTC)
+ date = current_date + timedelta(days=3)
+ matcher = DateIsInPast(current_date=current_date)
+ self.assertEqual(current_date, matcher.match(date).limit)
