launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #00375
Re: [Merge] lp:~james-w/launchpad/no-more-sampledata-1 into lp:launchpad/devel
On Sun, Aug 1, 2010 at 9:55 PM, James Westby <jw+debian@xxxxxxxxxxxxxxx> wrote:
> James Westby has proposed merging lp:~james-w/launchpad/no-more-sampledata-1 into lp:launchpad/devel with lp:~james-w/launchpad/no-more-sampledata-0 as a prerequisite.
>
> Requested reviews:
> Launchpad code reviewers (launchpad-reviewers)
>
>
> Hi,
>
> Here's some more cleanup of test_archive to remove reliance on
> sampledata (or at least SoyuzTestPublisher.setup_breezy_autotest()).
>
> There are a couple of fixes here too, one for bug 612351, and one
> where the build scoring code assumed it new the complete list of
> components, but they are stored in the db, not in an enum or similar.
>
Sweet, thanks.
I've got a few trivial suggestions, and would like to see some tests
for the changes made to the factory. Other than that, looks good.
I look forward to hearing back from you.
jml
> === modified file 'lib/lp/soyuz/model/buildpackagejob.py'
> --- lib/lp/soyuz/model/buildpackagejob.py 2010-06-10 12:18:10 +0000
> +++ lib/lp/soyuz/model/buildpackagejob.py 2010-08-01 20:54:49 +0000
> @@ -115,9 +115,10 @@
> score += score_pocket
>
> # Calculates the component-related part of the score.
> - score_component = score_componentname[
> - self.build.current_component.name]
> - score += score_component
> + if self.build.current_component.name in score_componentname:
> + score_component = score_componentname[
> + self.build.current_component.name]
> + score += score_component
>
FWIW, this can also be written as:
score_component =
score_componentname.get(self.build.current_component.name, 0)
score += score_component
Which I *think* might be a little faster. Whether it's clearer or not
is subjective. Your call as to whether you make the change.
> # Calculates the build queue time component of the score.
> right_now = datetime.now(pytz.timezone('UTC'))
>
> === modified file 'lib/lp/soyuz/tests/test_archive.py'
> --- lib/lp/soyuz/tests/test_archive.py 2010-08-01 20:54:47 +0000
> +++ lib/lp/soyuz/tests/test_archive.py 2010-08-01 20:54:49 +0000
> @@ -1,9 +1,11 @@
> # Copyright 2009-2010 Canonical Ltd. This software is licensed under the
> # GNU Affero General Public License version 3 (see the file LICENSE).
>
> +from __future__ import with_statement
> +
> """Test Archive features."""
>
> -from datetime import date, timedelta
> +from datetime import date
>
> import transaction
>
> @@ -12,18 +14,21 @@
> from zope.security.proxy import removeSecurityProxy
>
> from canonical.database.sqlbase import sqlvalues
> +from canonical.launchpad.interfaces.launchpad import ILaunchpadCelebrities
> from canonical.launchpad.webapp.interfaces import (
> IStoreSelector, MAIN_STORE, DEFAULT_FLAVOR)
> from canonical.testing import DatabaseFunctionalLayer, LaunchpadZopelessLayer
>
> from lp.buildmaster.interfaces.buildbase import BuildStatus
> -from lp.registry.interfaces.distribution import IDistributionSet
> -from lp.registry.interfaces.person import IPersonSet
> from lp.registry.interfaces.pocket import PackagePublishingPocket
> +from lp.registry.interfaces.series import SeriesStatus
> from lp.services.job.interfaces.job import JobStatus
> -from lp.soyuz.interfaces.archive import (IArchiveSet, ArchivePurpose,
> - ArchiveStatus, CannotRestrictArchitectures, CannotSwitchPrivacy,
> - InvalidPocketForPartnerArchive, InvalidPocketForPPA)
> +from lp.soyuz.interfaces.archive import (
> + ArchiveDisabled, ArchivePurpose, ArchiveStatus,
> + CannotRestrictArchitectures, CannotSwitchPrivacy, CannotUploadToPocket,
> + CannotUploadToPPA, IArchiveSet, InsufficientUploadRights,
> + InvalidPocketForPartnerArchive, InvalidPocketForPPA, NoRightsForArchive,
> + NoRightsForComponent)
> from lp.services.worlddata.interfaces.country import ICountrySet
> from lp.soyuz.interfaces.archivearch import IArchiveArchSet
> from lp.soyuz.interfaces.binarypackagename import IBinaryPackageNameSet
> @@ -32,11 +37,11 @@
> from lp.soyuz.interfaces.component import IComponentSet
> from lp.soyuz.interfaces.processor import IProcessorFamilySet
> from lp.soyuz.interfaces.publishing import PackagePublishingStatus
> -from lp.soyuz.model.binarypackagebuild import BinaryPackageBuild
> from lp.soyuz.model.binarypackagerelease import (
> BinaryPackageReleaseDownloadCount)
> from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
> -from lp.testing import login, login_person, TestCaseWithFactory
> +from lp.testing import (
> + login, login_person, person_logged_in, TestCaseWithFactory)
>
>
> class TestGetPublicationsInArchive(TestCaseWithFactory):
> @@ -203,247 +208,128 @@
> class TestSeriesWithSources(TestCaseWithFactory):
> """Create some sources in different series."""
>
> - layer = LaunchpadZopelessLayer
> -
> - def setUp(self):
> - super(TestSeriesWithSources, self).setUp()
> - self.publisher = SoyuzTestPublisher()
> - self.publisher.prepareBreezyAutotest()
> -
> - # Create three sources for the two different distroseries.
> - breezy_autotest = self.publisher.distroseries
> - ubuntu_test = breezy_autotest.distribution
> - self.series = [breezy_autotest]
> - self.series.append(self.factory.makeDistroRelease(
> - distribution=ubuntu_test, name="foo-series", version='1.0'))
> -
> - self.sources = []
> - gedit_src_hist = self.publisher.getPubSource(
> - sourcename="gedit", status=PackagePublishingStatus.PUBLISHED)
> - self.sources.append(gedit_src_hist)
> -
> - firefox_src_hist = self.publisher.getPubSource(
> - sourcename="firefox", status=PackagePublishingStatus.PUBLISHED,
> - distroseries=self.series[1])
> - self.sources.append(firefox_src_hist)
> -
> - gtg_src_hist = self.publisher.getPubSource(
> - sourcename="getting-things-gnome",
> - status=PackagePublishingStatus.PUBLISHED,
> - distroseries=self.series[1])
> - self.sources.append(gtg_src_hist)
> -
> - # Shortcuts for test readability.
> - self.archive = self.series[0].main_archive
> + layer = DatabaseFunctionalLayer
>
> def test_series_with_sources_returns_all_series(self):
> # Calling series_with_sources returns all series with publishings.
> - series = self.archive.series_with_sources
> - series_names = [s.displayname for s in series]
> -
> - self.assertContentEqual(
> - [u'Breezy Badger Autotest', u'Foo-series'],
> - series_names)
> + distribution = self.factory.makeDistribution()
> + archive = self.factory.makeArchive(distribution=distribution)
> + series_with_no_sources = self.factory.makeDistroSeries(
> + distribution=distribution)
> + series_with_sources1 = self.factory.makeDistroSeries(
> + distribution=distribution)
> + self.factory.makeSourcePackagePublishingHistory(
> + distroseries=series_with_sources1, archive=archive,
> + status=PackagePublishingStatus.PUBLISHED)
> + series_with_sources2 = self.factory.makeDistroSeries(
> + distribution=distribution)
> + self.factory.makeSourcePackagePublishingHistory(
> + distroseries=series_with_sources2, archive=archive,
> + status=PackagePublishingStatus.PENDING)
> + self.assertEqual(
> + [series_with_sources1, series_with_sources2],
> + sorted(archive.series_with_sources))
>
> def test_series_with_sources_ignore_non_published_records(self):
> # If all publishings in a series are deleted or superseded
> # the series will not be returned.
> - self.sources[0].status = (
> - PackagePublishingStatus.DELETED)
> -
> - series = self.archive.series_with_sources
> - series_names = [s.displayname for s in series]
> -
> - self.assertContentEqual([u'Foo-series'], series_names)
> + series = self.factory.makeDistroSeries()
> + archive = self.factory.makeArchive(distribution=series.distribution)
> + self.factory.makeSourcePackagePublishingHistory(
> + distroseries=series, archive=archive,
> + status=PackagePublishingStatus.DELETED)
> + self.assertEqual([], archive.series_with_sources)
>
> def test_series_with_sources_ordered_by_version(self):
> # The returned series are ordered by the distroseries version.
> - series = self.archive.series_with_sources
> - versions = [s.version for s in series]
> -
> - # Latest version should be first
> - self.assertEqual(
> - [u'6.6.6', u'1.0'], versions,
> - "The latest version was not first.")
> -
> - # Update the version of breezyautotest and ensure that the
> - # latest version is still first.
> - self.series[0].version = u'0.5'
> - series = self.archive.series_with_sources
> - versions = [s.version for s in series]
> - self.assertEqual(
> - [u'1.0', u'0.5'], versions,
> - "The latest version was not first.")
> + distribution = self.factory.makeDistribution()
> + archive = self.factory.makeArchive(distribution=distribution)
> + series1 = self.factory.makeDistroSeries(
> + version="1", distribution=distribution)
> + series2 = self.factory.makeDistroSeries(
> + version="2", distribution=distribution)
> + self.factory.makeSourcePackagePublishingHistory(
> + distroseries=series1, archive=archive,
> + status=PackagePublishingStatus.PUBLISHED)
> + self.factory.makeSourcePackagePublishingHistory(
> + distroseries=series2, archive=archive,
> + status=PackagePublishingStatus.PUBLISHED)
> + self.assertEqual([series2, series1], archive.series_with_sources)
> + # Change the version such that they should order differently
> + removeSecurityProxy(series2).version = "0.5"
> + # ... and check that they do
> + self.assertEqual([series1, series2], archive.series_with_sources)
>
>
> class TestGetSourcePackageReleases(TestCaseWithFactory):
>
> - layer = LaunchpadZopelessLayer
> -
> - def setUp(self):
> - super(TestGetSourcePackageReleases, self).setUp()
> - self.publisher = SoyuzTestPublisher()
> - self.publisher.prepareBreezyAutotest()
> -
> - # Create an archive with some published binaries.
> - self.archive = self.factory.makeArchive()
> - binaries_foo = self.publisher.getPubBinaries(
> - archive=self.archive, binaryname="foo-bin")
> - binaries_bar = self.publisher.getPubBinaries(
> - archive=self.archive, binaryname="bar-bin")
> -
> - # Collect the builds for reference.
> - self.builds_foo = [
> - binary.binarypackagerelease.build for binary in binaries_foo]
> - self.builds_bar = [
> - binary.binarypackagerelease.build for binary in binaries_bar]
> -
> - # Collect the source package releases for reference.
> - self.sourcepackagereleases = [
> - self.builds_foo[0].source_package_release,
> - self.builds_bar[0].source_package_release,
> - ]
> + layer = DatabaseFunctionalLayer
> +
> + def create_archive_with_builds(self, statuses):
Sadly, createArchiveWithBuilds is the preferred nomenclature.
> + archive = self.factory.makeArchive()
> + sprs = []
> + for status in statuses:
> + sourcepackagerelease = self.factory.makeSourcePackageRelease()
> + self.factory.makeBinaryPackageBuild(
> + source_package_release=sourcepackagerelease,
> + archive=archive, status=status)
> + sprs.append(sourcepackagerelease)
> + unlinked_spr = self.factory.makeSourcePackageRelease()
> + return archive, sprs
>
> def test_getSourcePackageReleases_with_no_params(self):
> # With no params all source package releases are returned.
> - sprs = self.archive.getSourcePackageReleases()
> -
> - self.assertContentEqual(self.sourcepackagereleases, sprs)
> + archive, sprs = self.create_archive_with_builds(
> + [BuildStatus.NEEDSBUILD, BuildStatus.FULLYBUILT])
> + self.assertContentEqual(
> + sprs, archive.getSourcePackageReleases())
>
> def test_getSourcePackageReleases_with_buildstatus(self):
> # Results are filtered by the specified buildstatus.
> -
> - # Set the builds for one of the sprs to needs build.
> - for build in self.builds_foo:
> - removeSecurityProxy(build).status = BuildStatus.NEEDSBUILD
> -
> - result = self.archive.getSourcePackageReleases(
> - build_status=BuildStatus.NEEDSBUILD)
> -
> - self.failUnlessEqual(1, result.count())
> - self.failUnlessEqual(
> - self.sourcepackagereleases[0], result[0])
> + archive, sprs = self.create_archive_with_builds(
> + [BuildStatus.NEEDSBUILD, BuildStatus.FULLYBUILT])
> + self.assertContentEqual(
> + [sprs[0]], archive.getSourcePackageReleases(
> + build_status=BuildStatus.NEEDSBUILD))
>
>
> class TestCorrespondingDebugArchive(TestCaseWithFactory):
>
> - layer = LaunchpadZopelessLayer
> -
> - def setUp(self):
> - super(TestCorrespondingDebugArchive, self).setUp()
> -
> - self.ubuntutest = getUtility(IDistributionSet)['ubuntutest']
> -
> - # Create a debug archive, as there isn't one in the sample data.
> - self.debug_archive = getUtility(IArchiveSet).new(
> - purpose=ArchivePurpose.DEBUG,
> - distribution=self.ubuntutest,
> - owner=self.ubuntutest.owner)
> -
> - # Retrieve sample data archives of each type.
> - self.primary_archive = getUtility(IArchiveSet).getByDistroPurpose(
> - self.ubuntutest, ArchivePurpose.PRIMARY)
> - self.partner_archive = getUtility(IArchiveSet).getByDistroPurpose(
> - self.ubuntutest, ArchivePurpose.PARTNER)
> - self.copy_archive = getUtility(IArchiveSet).getByDistroPurpose(
> - self.ubuntutest, ArchivePurpose.PARTNER)
> - self.ppa = getUtility(IPersonSet).getByName('cprov').archive
> + layer = DatabaseFunctionalLayer
>
> def testPrimaryDebugArchiveIsDebug(self):
> - self.assertEquals(
> - self.primary_archive.debug_archive, self.debug_archive)
> + distribution = self.factory.makeDistribution()
> + primary = self.factory.makeArchive(
> + distribution=distribution, purpose=ArchivePurpose.PRIMARY)
> + debug = self.factory.makeArchive(
> + distribution=distribution, purpose=ArchivePurpose.DEBUG)
> + self.assertEquals(primary.debug_archive, debug)
>
> def testPartnerDebugArchiveIsSelf(self):
> - self.assertEquals(
> - self.partner_archive.debug_archive, self.partner_archive)
> + partner = self.factory.makeArchive(purpose=ArchivePurpose.PARTNER)
> + self.assertEquals(partner.debug_archive, partner)
>
> def testCopyDebugArchiveIsSelf(self):
> - self.assertEquals(
> - self.copy_archive.debug_archive, self.copy_archive)
> + copy = self.factory.makeArchive(purpose=ArchivePurpose.COPY)
> + self.assertEquals(copy.debug_archive, copy)
>
> def testDebugDebugArchiveIsSelf(self):
> - self.assertEquals(
> - self.debug_archive.debug_archive, self.debug_archive)
> + debug = self.factory.makeArchive(purpose=ArchivePurpose.DEBUG)
> + self.assertEquals(debug.debug_archive, debug)
>
> def testPPADebugArchiveIsSelf(self):
> - self.assertEquals(self.ppa.debug_archive, self.ppa)
> + ppa = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
> + self.assertEquals(ppa.debug_archive, ppa)
>
> def testMissingPrimaryDebugArchiveIsNone(self):
> - # Turn the DEBUG archive into a COPY archive to hide it.
> - removeSecurityProxy(self.debug_archive).purpose = ArchivePurpose.COPY
> -
> - self.assertIs(
> - self.primary_archive.debug_archive, None)
> + primary = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY)
> + self.assertIs(primary.debug_archive, None)
>
>
> class TestArchiveEnableDisable(TestCaseWithFactory):
> """Test the enable and disable methods of Archive."""
>
> - layer = LaunchpadZopelessLayer
> -
> - def setUp(self):
> - #XXX: rockstar - 12 Jan 2010 - Bug #506255 - Tidy up these tests!
> - super(TestArchiveEnableDisable, self).setUp()
> -
> - self.publisher = SoyuzTestPublisher()
> - self.publisher.prepareBreezyAutotest()
> -
> - self.ubuntutest = getUtility(IDistributionSet)['ubuntutest']
> - self.archive = getUtility(IArchiveSet).getByDistroPurpose(
> - self.ubuntutest, ArchivePurpose.PRIMARY)
> -
> - store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
> - sample_data = store.find(BinaryPackageBuild)
> - for build in sample_data:
> - build.buildstate = BuildStatus.FULLYBUILT
> - store.flush()
> -
> - # We test builds that target a primary archive.
> - self.builds = []
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="gedit", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="firefox",
> - status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="apg", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="vim", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="gcc", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="bison", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="flex", status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - self.builds.extend(
> - self.publisher.getPubSource(
> - sourcename="postgres",
> - status=PackagePublishingStatus.PUBLISHED,
> - archive=self.archive).createMissingBuilds())
> - # Set up the builds for test.
> - score = 1000
> - duration = 0
> - for build in self.builds:
> - score += 1
> - duration += 60
> - bq = build.buildqueue_record
> - bq.lastscore = score
> - removeSecurityProxy(bq).estimated_duration = timedelta(
> - seconds=duration)
> + layer = DatabaseFunctionalLayer
>
> def _getBuildJobsByStatus(self, archive, status):
> # Return the count for archive build jobs with the given status.
> @@ -470,92 +356,95 @@
> # status.
> self.assertEqual(self._getBuildJobsByStatus(archive, status), 0)
>
> - def assertHasBuildJobsWithStatus(self, archive, status):
> + def assertHasBuildJobsWithStatus(self, archive, status, count):
> # Check that that there are jobs attached to this archive that have
> # the specified status.
> - self.assertEqual(self._getBuildJobsByStatus(archive, status), 8)
> + self.assertEqual(self._getBuildJobsByStatus(archive, status), count)
>
> def test_enableArchive(self):
> # Enabling an archive should set all the Archive's suspended builds to
> # WAITING.
> -
> - # Disable the archive, because it's currently enabled.
> - self.archive.disable()
> - self.assertHasBuildJobsWithStatus(self.archive, JobStatus.SUSPENDED)
> - self.archive.enable()
> - self.assertNoBuildJobsHaveStatus(self.archive, JobStatus.SUSPENDED)
> - self.assertTrue(self.archive.enabled)
> + archive = self.factory.makeArchive(enabled=True)
> + self.factory.makeBinaryPackageBuild(
> + archive=archive, status=BuildStatus.NEEDSBUILD)
> + # disable the archive, as it is currently enabled
> + removeSecurityProxy(archive).disable()
> + self.assertHasBuildJobsWithStatus(archive, JobStatus.SUSPENDED, 1)
> + removeSecurityProxy(archive).enable()
> + self.assertNoBuildJobsHaveStatus(archive, JobStatus.SUSPENDED)
> + self.assertTrue(archive.enabled)
>
> def test_enableArchiveAlreadyEnabled(self):
> # Enabling an already enabled Archive should raise an AssertionError.
> - self.assertRaises(AssertionError, self.archive.enable)
> + archive = self.factory.makeArchive(enabled=True)
> + self.assertRaises(AssertionError, removeSecurityProxy(archive).enable)
>
> def test_disableArchive(self):
> # Disabling an archive should set all the Archive's pending bulds to
> # SUSPENDED.
> - self.assertHasBuildJobsWithStatus(self.archive, JobStatus.WAITING)
> - self.archive.disable()
> - self.assertNoBuildJobsHaveStatus(self.archive, JobStatus.WAITING)
> - self.assertFalse(self.archive.enabled)
> + archive = self.factory.makeArchive(enabled=True)
> + self.factory.makeBinaryPackageBuild(
> + archive=archive, status=BuildStatus.NEEDSBUILD)
> + self.assertHasBuildJobsWithStatus(archive, JobStatus.WAITING, 1)
> + removeSecurityProxy(archive).disable()
> + self.assertNoBuildJobsHaveStatus(archive, JobStatus.WAITING)
> + self.assertFalse(archive.enabled)
>
> def test_disableArchiveAlreadyDisabled(self):
> # Disabling an already disabled Archive should raise an
> # AssertionError.
Ugh. In a better world, we'd be rich enough to afford our own exceptions.
No need to change anything.
> - self.archive.disable()
> - self.assertRaises(AssertionError, self.archive.disable)
> + archive = self.factory.makeArchive(enabled=False)
> + self.assertRaises(
> + AssertionError, removeSecurityProxy(archive).disable)
>
>
> class TestCollectLatestPublishedSources(TestCaseWithFactory):
> """Ensure that the private helper method works as expected."""
>
> - layer = LaunchpadZopelessLayer
> -
> - def setUp(self):
> - """Setup an archive with relevant publications."""
> - super(TestCollectLatestPublishedSources, self).setUp()
> - self.publisher = SoyuzTestPublisher()
> - self.publisher.prepareBreezyAutotest()
> -
> - # Create an archive with some published sources. We'll store
> - # a reference to the naked archive so that we can call
> - # the private method which is not defined on the interface.
> - self.archive = self.factory.makeArchive()
> - self.naked_archive = removeSecurityProxy(self.archive)
> -
> - self.pub_1 = self.publisher.getPubSource(
> - version='0.5.11~ppa1', archive=self.archive, sourcename="foo",
> - status=PackagePublishingStatus.PUBLISHED)
> -
> - self.pub_2 = self.publisher.getPubSource(
> - version='0.5.11~ppa2', archive=self.archive, sourcename="foo",
> - status=PackagePublishingStatus.PUBLISHED)
> -
> - self.pub_3 = self.publisher.getPubSource(
> - version='0.9', archive=self.archive, sourcename="bar",
> - status=PackagePublishingStatus.PUBLISHED)
> + layer = DatabaseFunctionalLayer
> +
> + def make_published_sources(self, archive, statuses, versions, names):
makePublishedSources.
> + for status, version, name in zip(statuses, versions, names):
> + self.factory.makeSourcePackagePublishingHistory(
> + sourcepackagename=name, archive=archive,
> + version=version, status=status)
>
> def test_collectLatestPublishedSources_returns_latest(self):
> - pubs = self.naked_archive._collectLatestPublishedSources(
> - self.archive, ["foo"])
> + sourcepackagename = self.factory.makeSourcePackageName(name="foo")
> + other_spn = self.factory.makeSourcePackageName(name="bar")
> + archive = self.factory.makeArchive()
> + self.make_published_sources(archive,
> + [PackagePublishingStatus.PUBLISHED]*3,
> + ["1.0", "1.1", "2.0"],
> + [sourcepackagename, sourcepackagename, other_spn])
> + pubs = removeSecurityProxy(archive)._collectLatestPublishedSources(
> + archive, ["foo"])
> self.assertEqual(1, len(pubs))
> - self.assertEqual('0.5.11~ppa2', pubs[0].source_package_version)
> + self.assertEqual('1.1', pubs[0].source_package_version)
>
> def test_collectLatestPublishedSources_returns_published_only(self):
> # Set the status of the latest pub to DELETED and ensure that it
> # is not returned.
> - self.pub_2.status = PackagePublishingStatus.DELETED
> -
> - pubs = self.naked_archive._collectLatestPublishedSources(
> - self.archive, ["foo"])
> + sourcepackagename = self.factory.makeSourcePackageName(name="foo")
> + other_spn = self.factory.makeSourcePackageName(name="bar")
> + archive = self.factory.makeArchive()
> + self.make_published_sources(archive,
> + [PackagePublishingStatus.PUBLISHED,
> + PackagePublishingStatus.DELETED,
> + PackagePublishingStatus.PUBLISHED],
> + ["1.0", "1.1", "2.0"],
> + [sourcepackagename, sourcepackagename, other_spn])
> + pubs = removeSecurityProxy(archive)._collectLatestPublishedSources(
> + archive, ["foo"])
> self.assertEqual(1, len(pubs))
> - self.assertEqual('0.5.11~ppa1', pubs[0].source_package_version)
> + self.assertEqual('1.0', pubs[0].source_package_version)
>
>
> class TestArchiveCanUpload(TestCaseWithFactory):
> """Test the various methods that verify whether uploads are allowed to
> happen."""
>
> - layer = LaunchpadZopelessLayer
> + layer = DatabaseFunctionalLayer
>
> def test_checkArchivePermission_by_PPA_owner(self):
> # Uploading to a PPA should be allowed for a user that is the owner
> @@ -568,50 +457,328 @@
>
> def test_checkArchivePermission_distro_archive(self):
> # Regular users can not upload to ubuntu
> - ubuntu = getUtility(IDistributionSet).getByName('ubuntu')
> - archive = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY,
> - distribution=ubuntu)
> + archive = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY)
> main = getUtility(IComponentSet)["main"]
> # A regular user doesn't have access
> - somebody = self.factory.makePerson(name="somebody")
> + somebody = self.factory.makePerson()
> self.assertEquals(False,
> archive.checkArchivePermission(somebody, main))
> # An ubuntu core developer does have access
> - kamion = getUtility(IPersonSet).getByName('kamion')
> - self.assertEquals(True, archive.checkArchivePermission(kamion, main))
> + coredev = self.factory.makePerson()
> + with person_logged_in(archive.owner):
> + archive.newComponentUploader(coredev, main.name)
> + self.assertEquals(True, archive.checkArchivePermission(coredev, main))
>
> def test_checkArchivePermission_ppa(self):
> - ubuntu = getUtility(IDistributionSet).getByName('ubuntu')
> - owner = self.factory.makePerson(name="eigenaar")
> + owner = self.factory.makePerson()
> archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA,
> - distribution=ubuntu,
> owner=owner)
> - somebody = self.factory.makePerson(name="somebody")
> + somebody = self.factory.makePerson()
> # The owner has access
> self.assertEquals(True, archive.checkArchivePermission(owner))
> # Somebody unrelated does not
> self.assertEquals(False, archive.checkArchivePermission(somebody))
>
> + def make_archive_and_active_distroseries(self, purpose=None):
makeArchiveAndActiveDistroSeries
> + if purpose is None:
> + purpose = ArchivePurpose.PRIMARY
> + archive = self.factory.makeArchive(purpose=purpose)
> + distroseries = self.factory.makeDistroSeries(
> + distribution=archive.distribution,
> + status=SeriesStatus.DEVELOPMENT)
> + return archive, distroseries
> +
> + def make_person_with_component_permission(self, archive):
makePersonWithComponentPermission
> + person = self.factory.makePerson()
> + component = self.factory.makeComponent()
> + removeSecurityProxy(archive).newComponentUploader(
> + person, component)
> + return person, component
> +
> + def checkUpload(self, archive, person=None, distroseries=None,
> + sourcepackagename=None, component=None,
> + pocket=None, strict_component=False):
> + if person is None:
> + person = self.factory.makePerson()
> + if distroseries is None:
> + distroseries = self.factory.makeDistroSeries()
> + if sourcepackagename is None:
> + sourcepackagename = self.factory.makeSourcePackageName()
> + if component is None:
> + component = self.factory.makeComponent()
> + if pocket is None:
> + pocket = PackagePublishingPocket.PROPOSED
> + return archive.checkUpload(
> + person, distroseries, sourcepackagename, component, pocket,
> + strict_component=strict_component)
> +
> def test_checkUpload_partner_invalid_pocket(self):
> # Partner archives only have release and proposed pockets
> - archive = self.factory.makeArchive(purpose=ArchivePurpose.PARTNER)
> - self.assertIsInstance(archive.checkUpload(self.factory.makePerson(),
> - self.factory.makeDistroSeries(),
> - self.factory.makeSourcePackageName(),
> - self.factory.makeComponent(),
> - PackagePublishingPocket.UPDATES),
> - InvalidPocketForPartnerArchive)
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PARTNER)
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, distroseries=distroseries,
> + pocket=PackagePublishingPocket.UPDATES),
> + InvalidPocketForPartnerArchive)
>
> def test_checkUpload_ppa_invalid_pocket(self):
> # PPA archives only have release pockets
> - archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
> - self.assertIsInstance(archive.checkUpload(self.factory.makePerson(),
> - self.factory.makeDistroSeries(),
> - self.factory.makeSourcePackageName(),
> - self.factory.makeComponent(),
> - PackagePublishingPocket.PROPOSED),
> - InvalidPocketForPPA)
> - # XXX: JRV 20100511: IArchive.canUploadSuiteSourcePackage needs tests
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PPA)
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, distroseries=distroseries,
> + pocket=PackagePublishingPocket.PROPOSED),
> + InvalidPocketForPPA)
> +
> + def test_checkUpload_invalid_pocket_for_series_state(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, distroseries=distroseries,
> + pocket=PackagePublishingPocket.PROPOSED),
> + CannotUploadToPocket)
> +
> + def test_checkUpload_disabled_archive(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + removeSecurityProxy(archive).disable()
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, distroseries=distroseries,
> + pocket=PackagePublishingPocket.RELEASE),
> + ArchiveDisabled)
> +
> + def test_checkUpload_ppa_owner(self):
> + person = self.factory.makePerson()
> + archive = self.factory.makeArchive(
> + purpose=ArchivePurpose.PPA, owner=person)
> + self.assertEqual(
> + None,
> + self.checkUpload(
> + archive, person=person,
> + pocket=PackagePublishingPocket.RELEASE))
> +
> + def test_checkUpload_ppa_with_permission(self):
> + archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newComponentUploader(person, "main")
> + # component is ignored
> + self.assertEqual(
> + None,
> + self.checkUpload(
> + archive, person=person,
> + component=self.factory.makeComponent(name="universe"),
> + pocket=PackagePublishingPocket.RELEASE))
> +
> + def test_checkUpload_ppa_with_no_permission(self):
> + archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
> + person = self.factory.makePerson()
> + self.assertIsInstance(
> + archive.checkUpload(
> + person, self.factory.makeDistroSeries(),
> + self.factory.makeSourcePackageName(),
> + self.factory.makeComponent(),
> + PackagePublishingPocket.RELEASE),
> + CannotUploadToPPA)
> +
> + def test_checkUpload_copy_archive_no_permission(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.COPY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newPackageUploader(
> + person, sourcepackagename)
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + pocket=PackagePublishingPocket.RELEASE),
> + NoRightsForArchive)
> +
> + def test_checkUpload_package_permission(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newPackageUploader(
> + person, sourcepackagename)
> + self.assertEquals(
> + None,
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + pocket=PackagePublishingPocket.RELEASE))
> +
> + def make_person_with_packageset_permission(self, archive, distroseries,
> + packages=()):
> + packageset = self.factory.makePackageset(
> + distroseries=distroseries, packages=packages)
> + person = self.factory.makePerson()
> + techboard = getUtility(ILaunchpadCelebrities).ubuntu_techboard
> + with person_logged_in(techboard):
> + archive.newPackagesetUploader(person, packageset)
> + return person, packageset
> +
> + def test_checkUpload_packageset_permission(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person, packageset = self.make_person_with_packageset_permission(
> + archive, distroseries, packages=[sourcepackagename])
> + self.assertEquals(
> + None,
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + pocket=PackagePublishingPocket.RELEASE))
> +
> + def test_checkUpload_component_permission(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person, component = self.make_person_with_component_permission(
> + archive)
> + self.assertEquals(
> + None,
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename, component=component,
> + pocket=PackagePublishingPocket.RELEASE))
> +
> + def test_checkUpload_no_permissions(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person = self.factory.makePerson()
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + pocket=PackagePublishingPocket.RELEASE),
> + NoRightsForArchive)
> +
> + def test_checkUpload_insufficient_permissions(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person, packageset = self.make_person_with_packageset_permission(
> + archive, distroseries)
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + pocket=PackagePublishingPocket.RELEASE),
> + InsufficientUploadRights)
> +
> + def test_checkUpload_without_strict_component(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person, component = self.make_person_with_component_permission(
> + archive)
> + other_component = self.factory.makeComponent()
> + self.assertEquals(
> + None,
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + component=other_component,
> + pocket=PackagePublishingPocket.RELEASE,
> + strict_component=False))
> +
> + def test_checkUpload_with_strict_component(self):
> + archive, distroseries = self.make_archive_and_active_distroseries(
> + purpose=ArchivePurpose.PRIMARY)
> + sourcepackagename = self.factory.makeSourcePackageName()
> + person, component = self.make_person_with_component_permission(
> + archive)
> + other_component = self.factory.makeComponent()
> + self.assertIsInstance(
> + self.checkUpload(
> + archive, person=person, distroseries=distroseries,
> + sourcepackagename=sourcepackagename,
> + component=other_component,
> + pocket=PackagePublishingPocket.RELEASE,
> + strict_component=True),
> + NoRightsForComponent)
> +
> + def make_package_to_upload(self, distroseries):
makePackageToUpload
> + sourcepackagename = self.factory.makeSourcePackageName()
> + suitesourcepackage = self.factory.makeSuiteSourcePackage(
> + pocket=PackagePublishingPocket.RELEASE,
> + sourcepackagename=sourcepackagename,
> + distroseries=distroseries)
> + return suitesourcepackage
> +
> + def test_canUploadSuiteSourcePackage_invalid_pocket(self):
> + # Test that canUploadSuiteSourcePackage calls checkUpload for
> + # the pocket checks.
> + person = self.factory.makePerson()
> + archive = self.factory.makeArchive(
> + purpose=ArchivePurpose.PPA, owner=person)
> + suitesourcepackage = self.factory.makeSuiteSourcePackage(
> + pocket=PackagePublishingPocket.PROPOSED)
> + self.assertEqual(
> + False,
> + archive.canUploadSuiteSourcePackage(person, suitesourcepackage))
> +
> + def test_canUploadSuiteSourcePackage_no_permission(self):
> + # Test that canUploadSuiteSourcePackage calls verifyUpload for
> + # the permission checks.
> + archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
> + suitesourcepackage = self.factory.makeSuiteSourcePackage(
> + pocket=PackagePublishingPocket.RELEASE)
> + person = self.factory.makePerson()
> + self.assertEqual(
> + False,
> + archive.canUploadSuiteSourcePackage(person, suitesourcepackage))
> +
> + def test_canUploadSuiteSourcePackage_package_permission(self):
> + # Test that a package permission is enough to upload a new
> + # package.
> + archive, distroseries = self.make_archive_and_active_distroseries()
> + suitesourcepackage = self.make_package_to_upload(distroseries)
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newPackageUploader(
> + person, suitesourcepackage.sourcepackagename)
> + self.assertEqual(
> + True,
> + archive.canUploadSuiteSourcePackage(person, suitesourcepackage))
> +
> + def test_canUploadSuiteSourcePackage_component_permission(self):
> + # Test that component upload permission is enough to be
> + # allowed to upload a new package.
> + archive, distroseries = self.make_archive_and_active_distroseries()
> + suitesourcepackage = self.make_package_to_upload(distroseries)
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newComponentUploader(person, "universe")
> + self.assertEqual(
> + True,
> + archive.canUploadSuiteSourcePackage(person, suitesourcepackage))
> +
> + def test_canUploadSuiteSourcePackage_strict_component(self):
> + # Test that canUploadSuiteSourcePackage uses strict component
> + # checking.
> + archive, distroseries = self.make_archive_and_active_distroseries()
> + suitesourcepackage = self.make_package_to_upload(distroseries)
> + main_component = self.factory.makeComponent(name="main")
> + self.factory.makeSourcePackagePublishingHistory(
> + archive=archive, distroseries=distroseries,
> + sourcepackagename=suitesourcepackage.sourcepackagename,
> + status=PackagePublishingStatus.PUBLISHED,
> + pocket=PackagePublishingPocket.RELEASE,
> + component=main_component)
> + person = self.factory.makePerson()
> + removeSecurityProxy(archive).newComponentUploader(person, "universe")
> + # This time the user can't upload as there has been a
> + # publication and they don't have permission for the component
> + # the package is published in.
> + self.assertEqual(
> + False,
> + archive.canUploadSuiteSourcePackage(person, suitesourcepackage))
>
>
> class TestUpdatePackageDownloadCount(TestCaseWithFactory):
>
> === modified file 'lib/lp/testing/factory.py'
> --- lib/lp/testing/factory.py 2010-08-01 20:54:47 +0000
> +++ lib/lp/testing/factory.py 2010-08-01 20:54:49 +0000
> @@ -2353,7 +2353,8 @@
> libraryfile=libraryfile, filetype=filetype))
>
> def makeBinaryPackageBuild(self, source_package_release=None,
> - distroarchseries=None, archive=None, builder=None):
> + distroarchseries=None, archive=None, builder=None,
> + status=None):
Could you please add a test for this change? No need to test all the
bits, just the new parameter.
> """Create a BinaryPackageBuild.
>
> If archive is not supplied, the source_package_release is used
> @@ -2363,6 +2364,7 @@
> :param distroarchseries: The DistroArchSeries to use.
> :param archive: The Archive to use.
> :param builder: An optional builder to assign.
> + :param status: The BuildStatus for the build.
> """
> if archive is None:
> if source_package_release is None:
> @@ -2378,11 +2380,13 @@
> distroarchseries = self.makeDistroArchSeries(
> distroseries=source_package_release.upload_distroseries,
> processorfamily=processor.family)
> + if status is None:
> + status = BuildStatus.NEEDSBUILD
> binary_package_build = getUtility(IBinaryPackageBuildSet).new(
> source_package_release=source_package_release,
> processor=processor,
> distro_arch_series=distroarchseries,
> - status=BuildStatus.NEEDSBUILD,
> + status=status,
> archive=archive,
> pocket=PackagePublishingPocket.RELEASE,
> date_created=self.getUniqueDate())
> @@ -2601,7 +2605,8 @@
> isinstance(sourcepackagename, basestring)):
> sourcepackagename = self.getOrMakeSourcePackageName(
> sourcepackagename)
> - return SuiteSourcePackage(distroseries, pocket, sourcepackagename)
> + return ProxyFactory(
> + SuiteSourcePackage(distroseries, pocket, sourcepackagename))
>
Not sure how you should test this, but I feel obliged to make you
think about it :)
jml
--
https://code.launchpad.net/~james-w/launchpad/no-more-sampledata-1/+merge/31493
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~james-w/launchpad/no-more-sampledata-1 into lp:launchpad/devel.
References
