launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #00368
[Merge] lp:~james-w/launchpad/archive-collection into lp:launchpad/devel
James Westby has proposed merging lp:~james-w/launchpad/archive-collection into lp:launchpad/devel.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
Hi,
Here's a branch to add an ArchiveCollection based
on the new generic Collection.
It's fairly standalone right now, though I ported a
few easy queries in ArchiveSet methods over to it.
Hopefully once it is available it will see more use.
Thanks,
James
--
https://code.launchpad.net/~james-w/launchpad/archive-collection/+merge/31499
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~james-w/launchpad/archive-collection into lp:launchpad/devel.
=== modified file 'lib/lp/services/database/collection.py'
--- lib/lp/services/database/collection.py 2010-07-16 14:10:18 +0000
+++ lib/lp/services/database/collection.py 2010-08-02 00:41:06 +0000
@@ -68,9 +68,6 @@
base_tables = list(base.tables)
self.store = kwargs.get('store')
- if self.store is None:
- self.store = getUtility(IStoreSelector).get(
- MAIN_STORE, DEFAULT_FLAVOR)
self.tables = (
starting_tables + base_tables +
@@ -118,6 +115,9 @@
If no values are requested, this selects the type of object that
the Collection is a collection of.
"""
+ if self.store is None:
+ self.store = getUtility(IStoreSelector).get(
+ MAIN_STORE, DEFAULT_FLAVOR)
if len(self.tables) == 0:
source = self.store
else:
=== modified file 'lib/lp/soyuz/configure.zcml'
--- lib/lp/soyuz/configure.zcml 2010-06-30 17:35:36 +0000
+++ lib/lp/soyuz/configure.zcml 2010-08-02 00:41:06 +0000
@@ -67,6 +67,19 @@
interface="lp.soyuz.interfaces.archivearch.IArchiveArchSet"/>
</securedutility>
+ <!-- ArchiveCollection -->
+ <securedutility
+ class="lp.soyuz.model.archivecollection.ArchiveCollection"
+ provides="lp.soyuz.interfaces.archivecollection.IAllArchives">
+ <allow
+ interface="lp.soyuz.interfaces.archivecollection.IAllArchives" />
+ </securedutility>
+
+ <class class="lp.soyuz.model.archivecollection.ArchiveCollection">
+ <allow
+ interface="lp.soyuz.interfaces.archivecollection.IAllArchives" />
+ </class>
+
<!-- DistroSeriesPackageCache -->
<class
class="canonical.launchpad.database.DistroSeriesPackageCache">
=== added file 'lib/lp/soyuz/interfaces/archivecollection.py'
--- lib/lp/soyuz/interfaces/archivecollection.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/interfaces/archivecollection.py 2010-08-02 00:41:06 +0000
@@ -0,0 +1,70 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""A collection of archives.
+
+See `IArchiveCollection` for more details.
+"""
+
+__metaclass__ = type
+__all__ = [
+ 'IAllArchives',
+ 'IArchiveCollection',
+ ]
+
+from zope.interface import Interface
+
+
+class IArchiveCollection(Interface):
+ """A collection of archives.
+
+ An `IArchiveCollection` is an immutable collection of archives.
+
+ You can use the methods it provides to get a new collection with
+ a filtered set of archives.
+
+ Finally you can call `select` to get a ResultSet from the collection.
+ """
+
+ def select(*args):
+ """Return a ResultSet for this collection, with values set to args."""
+
+ def withPurpose(purpose):
+ """Restrict the archives to only the given purpose."""
+
+ def withPurposes(purposes):
+ """Restrict the archives to one of the given purposes."""
+
+ def withStatus(status):
+ """Restrict the archives to only the given status."""
+
+ def withDistribution(distribution):
+ """Restrict the archives to only the given distribution."""
+
+ def ownedBy(owner):
+ """Restrict the archives to only those owned by the given person."""
+
+ def withName(name):
+ """Restrict the archives to only those with the given name."""
+
+ def isEnabled(enabled=True):
+ """Restrict the archives to only those enabled as passed."""
+
+ def isPrivate(private=True):
+ """Restrict the archives to only those with the given privacy."""
+
+ def isRequireVirtualized(require_virtualized=True):
+ """Restrict the archives to those requiring virtualization."""
+
+ def isCommercial(commercial=True):
+ """Restrict the archives to commercial as given."""
+
+ def visibleByUser(user):
+ """Restrict the archives to those visible by the given user.
+
+ :param user: a Person, or None for public archives.
+ """
+
+
+class IAllArchives(IArchiveCollection):
+ """An `IArchiveCollection` representing all archives in Launchpad."""
=== modified file 'lib/lp/soyuz/model/archive.py'
--- lib/lp/soyuz/model/archive.py 2010-07-27 12:53:24 +0000
+++ lib/lp/soyuz/model/archive.py 2010-08-02 00:41:06 +0000
@@ -40,6 +40,7 @@
from lp.soyuz.adapters.packagelocation import PackageLocation
from canonical.launchpad.components.tokens import (
create_unique_token_for_table)
+from lp.soyuz.interfaces.archivecollection import IAllArchives
from lp.soyuz.model.archivedependency import ArchiveDependency
from lp.soyuz.model.archiveauthtoken import ArchiveAuthToken
from lp.soyuz.model.archivesubscriber import ArchiveSubscriber
@@ -1810,13 +1811,10 @@
def getPPAOwnedByPerson(self, person, name=None):
"""See `IArchiveSet`."""
- store = Store.of(person)
- clause = [
- Archive.purpose == ArchivePurpose.PPA,
- Archive.owner == person]
+ ppas = getUtility(IAllArchives).withPurpose(ArchivePurpose.PPA)
if name is not None:
- clause.append(Archive.name == name)
- result = store.find(Archive, *clause).order_by(Archive.id).first()
+ ppas = ppas.withName(name)
+ result = ppas.ownedBy(person).select().order_by(Archive.id).first()
if name is not None and result is None:
raise NoSuchPPA(name)
return result
@@ -1954,79 +1952,34 @@
def getPrivatePPAs(self):
"""See `IArchiveSet`."""
- store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
- return store.find(
- Archive,
- Archive.private == True,
- Archive.purpose == ArchivePurpose.PPA)
+ ppas = getUtility(IAllArchives).withPurpose(ArchivePurpose.PPA)
+ return ppas.isPrivate().select()
def getCommercialPPAs(self):
"""See `IArchiveSet`."""
- store = IStore(Archive)
- return store.find(
- Archive,
- Archive.commercial == True,
- Archive.purpose == ArchivePurpose.PPA)
+ ppas = getUtility(IAllArchives).withPurpose(ArchivePurpose.PPA)
+ return ppas.isCommercial().select()
def getArchivesForDistribution(self, distribution, name=None,
purposes=None, user=None,
exclude_disabled=True):
"""See `IArchiveSet`."""
- extra_exprs = []
+ archives = getUtility(IAllArchives).withDistribution(distribution)
- # If a single purpose is passed in, convert it into a tuple,
- # otherwise assume a list was passed in.
if purposes in ArchivePurpose:
- purposes = (purposes,)
-
- if purposes:
- extra_exprs.append(Archive.purpose.is_in(purposes))
+ archives = archives.withPurpose(purposes)
+ elif purposes:
+ archives = archives.withPurposes(*purposes)
if name is not None:
- extra_exprs.append(Archive.name == name)
-
- public_archive = And(Archive.private == False,
- Archive._enabled == True)
-
- if user is not None:
- admins = getUtility(ILaunchpadCelebrities).admin
- if not user.inTeam(admins):
- # Enforce privacy-awareness for logged-in, non-admin users,
- # so that they can only see the private archives that they're
- # allowed to see.
-
- # Create a subselect to capture all the teams that are
- # owners of archives AND the user is a member of:
- user_teams_subselect = Select(
- TeamParticipation.teamID,
- where=And(
- TeamParticipation.personID == user.id,
- TeamParticipation.teamID == Archive.ownerID))
-
- # Append the extra expression to capture either public
- # archives, or archives owned by the user, or archives
- # owned by a team of which the user is a member:
- # Note: 'Archive.ownerID == user.id'
- # is unnecessary below because there is a TeamParticipation
- # entry showing that each person is a member of the "team"
- # that consists of themselves.
-
- # FIXME: Include private PPA's if user is an uploader
- extra_exprs.append(
- Or(public_archive,
- Archive.ownerID.is_in(user_teams_subselect)))
- else:
- # Anonymous user; filter to include only public archives in
- # the results.
- extra_exprs.append(public_archive)
+ archives = archives.withName(name)
+
+ archives = archives.visibleByUser(user)
if exclude_disabled:
- extra_exprs.append(Archive._enabled == True)
+ archives = archives.isEnabled()
- query = Store.of(distribution).find(
- Archive,
- Archive.distribution == distribution,
- *extra_exprs)
+ query = archives.select()
return query.order_by(Archive.name)
=== added file 'lib/lp/soyuz/model/archivecollection.py'
--- lib/lp/soyuz/model/archivecollection.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/model/archivecollection.py 2010-08-02 00:41:06 +0000
@@ -0,0 +1,86 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+from storm.expr import And, Or, Select
+from zope.component import getUtility
+from zope.interface import implements
+
+from canonical.launchpad.interfaces.launchpad import ILaunchpadCelebrities
+from lp.registry.model.teammembership import TeamParticipation
+from lp.services.database.collection import Collection
+from lp.soyuz.interfaces.archivecollection import IArchiveCollection
+from lp.soyuz.model.archive import Archive
+
+
+class ArchiveCollection(Collection):
+
+ implements(IArchiveCollection)
+
+ starting_table = Archive
+
+ def withPurpose(self, purpose):
+ return self.refine(Archive.purpose == purpose)
+
+ def withPurposes(self, *purposes):
+ return self.refine(Archive.purpose.is_in(purposes))
+
+ def withStatus(self, status):
+ return self.refine(Archive.status == status)
+
+ def withDistribution(self, distribution):
+ return self.refine(Archive.distribution == distribution)
+
+ def ownedBy(self, owner):
+ return self.refine(Archive.owner == owner)
+
+ def withName(self, name):
+ return self.refine(Archive.name == name)
+
+ def isEnabled(self, enabled=True):
+ return self.refine(Archive._enabled == enabled)
+
+ def isPrivate(self, private=True):
+ return self.refine(Archive.private == private)
+
+ def isRequireVirtualized(self, require_virtualized=True):
+ return self.refine(Archive.require_virtualized == require_virtualized)
+
+ def isCommercial(self, commercial=True):
+ return self.refine(Archive.commercial == commercial)
+
+ def visibleByUser(self, user):
+ public_archive = And(Archive.private == False,
+ Archive._enabled == True)
+ if user is None:
+ # Anonymous user; filter to include only public archives in
+ # the results.
+ return self.refine(public_archive)
+ celebs = getUtility(ILaunchpadCelebrities)
+ if user.inTeam(celebs.admin):
+ # Admins can see everything
+ return self
+ # Enforce privacy-awareness for non-admin users,
+ # so that they can only see the private archives that they're
+ # allowed to see.
+
+ # Create a subselect to capture all the teams that are
+ # owners of archives AND the user is a member of:
+ user_teams_subselect = Select(
+ TeamParticipation.teamID,
+ where=And(
+ TeamParticipation.personID == user.id,
+ TeamParticipation.teamID == Archive.ownerID))
+
+ # Append the extra expression to capture either public
+ # archives, or archives owned by the user, or archives
+ # owned by a team of which the user is a member:
+ # Note: 'Archive.ownerID == user.id'
+ # is unnecessary below because there is a TeamParticipation
+ # entry showing that each person is a member of the "team"
+ # that consists of themselves.
+
+ # FIXME: Include private PPA's if user is an uploader
+ return self.refine(
+ Or(public_archive, Archive.ownerID.is_in(user_teams_subselect)))
=== added file 'lib/lp/soyuz/tests/test_archivecollection.py'
--- lib/lp/soyuz/tests/test_archivecollection.py 1970-01-01 00:00:00 +0000
+++ lib/lp/soyuz/tests/test_archivecollection.py 2010-08-02 00:41:06 +0000
@@ -0,0 +1,220 @@
+# Copyright 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for `ArchiveCollection`."""
+
+__metaclass__ = type
+
+from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
+
+from canonical.launchpad.interfaces.launchpad import ILaunchpadCelebrities
+from canonical.testing import DatabaseFunctionalLayer
+from lp.soyuz.interfaces.archive import ArchivePurpose, ArchiveStatus
+from lp.soyuz.model.archivecollection import ArchiveCollection
+from lp.testing import TestCaseWithFactory
+
+
+class TestGetPublicationsInArchive(TestCaseWithFactory):
+
+ layer = DatabaseFunctionalLayer
+
+ def test_baseline(self):
+ archive = self.factory.makeArchive()
+ collection = ArchiveCollection()
+ self.assertIn(archive, collection.select())
+
+ def test_purpose_found(self):
+ archive = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
+ collection = ArchiveCollection()
+ ppas = collection.withPurpose(ArchivePurpose.PPA)
+ self.assertIn(archive, ppas.select())
+
+ def test_purpose_not_found(self):
+ archive = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY)
+ collection = ArchiveCollection()
+ ppas = collection.withPurpose(ArchivePurpose.PPA)
+ self.assertNotIn(archive, ppas.select())
+
+ def test_purposes(self):
+ archive = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY)
+ ppa = self.factory.makeArchive(purpose=ArchivePurpose.PPA)
+ copy_archive = self.factory.makeArchive(purpose=ArchivePurpose.COPY)
+ archives = ArchiveCollection().withPurposes(
+ ArchivePurpose.PPA, ArchivePurpose.COPY)
+ result = archives.select()
+ self.assertIn(ppa, result)
+ self.assertIn(copy_archive, result)
+ self.assertNotIn(archive, result)
+
+ def test_distribution_found(self):
+ distribution = self.factory.makeDistribution()
+ archive = self.factory.makeArchive(distribution=distribution)
+ archives = ArchiveCollection().withDistribution(distribution)
+ self.assertIn(archive, archives.select())
+
+ def test_distribution_notfound(self):
+ distribution = self.factory.makeDistribution()
+ archive = self.factory.makeArchive(distribution=distribution)
+ other_distribution = self.factory.makeDistribution()
+ archives = ArchiveCollection().withDistribution(other_distribution)
+ self.assertNotIn(archive, archives.select())
+
+ def test_owner_found(self):
+ owner = self.factory.makePerson()
+ archive = self.factory.makeArchive(owner=owner)
+ archives = ArchiveCollection().ownedBy(owner)
+ self.assertIn(archive, archives.select())
+
+ def test_owner_not_found(self):
+ owner = self.factory.makePerson()
+ archive = self.factory.makeArchive(owner=owner)
+ other_person = self.factory.makePerson()
+ archives = ArchiveCollection().ownedBy(other_person)
+ self.assertNotIn(archive, archives.select())
+
+ def test_name_found(self):
+ name = "foo"
+ archive = self.factory.makeArchive(name=name)
+ archives = ArchiveCollection().withName(name)
+ self.assertIn(archive, archives.select())
+
+ def test_name_not_found(self):
+ name = "foo"
+ archive = self.factory.makeArchive(name=name)
+ archives = ArchiveCollection().withName("bar")
+ self.assertNotIn(archive, archives.select())
+
+ def test_enabled_found(self):
+ archive = self.factory.makeArchive(enabled=True)
+ archives = ArchiveCollection().isEnabled()
+ self.assertIn(archive, archives.select())
+
+ def test_enabled_not_found(self):
+ archive = self.factory.makeArchive(enabled=True)
+ archives = ArchiveCollection().isEnabled(False)
+ self.assertNotIn(archive, archives.select())
+
+ def test_disabled_found(self):
+ archive = self.factory.makeArchive(enabled=False)
+ archives = ArchiveCollection().isEnabled(False)
+ self.assertIn(archive, archives.select())
+
+ def test_disabled_not_found(self):
+ archive = self.factory.makeArchive(enabled=False)
+ archives = ArchiveCollection().isEnabled()
+ self.assertNotIn(archive, archives.select())
+
+ def test_private_found(self):
+ archive = self.factory.makeArchive(private=True)
+ archives = ArchiveCollection().isPrivate()
+ self.assertIn(archive, archives.select())
+
+ def test_private_not_found(self):
+ archive = self.factory.makeArchive(private=True)
+ archives = ArchiveCollection().isPrivate(False)
+ self.assertNotIn(archive, archives.select())
+
+ def test_non_private_found(self):
+ archive = self.factory.makeArchive(private=False)
+ archives = ArchiveCollection().isPrivate(False)
+ self.assertIn(archive, archives.select())
+
+ def test_non_private_not_found(self):
+ archive = self.factory.makeArchive(private=False)
+ archives = ArchiveCollection().isPrivate()
+ self.assertNotIn(archive, archives.select())
+
+ def test_require_virtualized_found(self):
+ archive = self.factory.makeArchive(virtualized=True)
+ archives = ArchiveCollection().isRequireVirtualized()
+ self.assertIn(archive, archives.select())
+
+ def test_require_virtualized_not_found(self):
+ archive = self.factory.makeArchive(virtualized=True)
+ archives = ArchiveCollection().isRequireVirtualized(False)
+ self.assertNotIn(archive, archives.select())
+
+ def test_not_require_virtualized_found(self):
+ archive = self.factory.makeArchive(virtualized=False)
+ archives = ArchiveCollection().isRequireVirtualized(False)
+ self.assertIn(archive, archives.select())
+
+ def test_not_require_virtualized_not_found(self):
+ archive = self.factory.makeArchive(virtualized=False)
+ archives = ArchiveCollection().isRequireVirtualized()
+ self.assertNotIn(archive, archives.select())
+
+ def test_status_found(self):
+ archive = self.factory.makeArchive(status=ArchiveStatus.ACTIVE)
+ archives = ArchiveCollection().withStatus(ArchiveStatus.ACTIVE)
+ self.assertIn(archive, archives.select())
+
+ def test_status_not_found(self):
+ archive = self.factory.makeArchive(status=ArchiveStatus.ACTIVE)
+ archives = ArchiveCollection().withStatus(ArchiveStatus.DELETING)
+ self.assertNotIn(archive, archives.select())
+
+ def test_commercial_found(self):
+ archive = self.factory.makeArchive(commercial=True)
+ archives = ArchiveCollection().isCommercial()
+ self.assertIn(archive, archives.select())
+
+ def test_commercial_not_found(self):
+ archive = self.factory.makeArchive(commercial=True)
+ archives = ArchiveCollection().isCommercial(False)
+ self.assertNotIn(archive, archives.select())
+
+ def test_not_commercial_not_found(self):
+ archive = self.factory.makeArchive(commercial=False)
+ archives = ArchiveCollection().isCommercial()
+ self.assertNotIn(archive, archives.select())
+
+ def test_not_commercial_found(self):
+ archive = self.factory.makeArchive(commercial=False)
+ archives = ArchiveCollection().isCommercial(False)
+ self.assertIn(archive, archives.select())
+
+ def test_admins_can_see_all(self):
+ archive = self.factory.makeArchive(private=True)
+ all_archives = ArchiveCollection()
+ admin = self.factory.makePerson()
+ admin_team = removeSecurityProxy(
+ getUtility(ILaunchpadCelebrities).admin)
+ admin_team.addMember(admin, admin_team.teamowner)
+ archives = ArchiveCollection().visibleByUser(admin)
+ self.assertEqual(
+ sorted(all_archives.select()), sorted(archives.select()))
+
+ def test_plain_user_cant_see_private(self):
+ archive = self.factory.makeArchive(private=True)
+ person = self.factory.makePerson()
+ archives = ArchiveCollection().visibleByUser(person)
+ self.assertNotIn(archive, archives.select())
+
+ def test_plain_user_cant_see_disabled(self):
+ archive = self.factory.makeArchive(enabled=False, private=False)
+ person = self.factory.makePerson()
+ archives = ArchiveCollection().visibleByUser(person)
+ self.assertNotIn(archive, archives.select())
+
+ def test_plain_user_can_see_public_enabled(self):
+ archive = self.factory.makeArchive(enabled=True, private=False)
+ person = self.factory.makePerson()
+ archives = ArchiveCollection().visibleByUser(person)
+ self.assertIn(archive, archives.select())
+
+ def test_user_can_see_private_disabled_archives_they_own(self):
+ person = self.factory.makePerson()
+ archive = self.factory.makeArchive(
+ enabled=False, private=True, owner=person)
+ archives = ArchiveCollection().visibleByUser(person)
+ self.assertIn(archive, archives.select())
+
+ def test_user_can_see_private_disabled_archives_for_their_teams(self):
+ person = self.factory.makePerson()
+ team = self.factory.makeTeam(members=[person])
+ archive = self.factory.makeArchive(
+ enabled=False, private=True, owner=team)
+ archives = ArchiveCollection().visibleByUser(person)
+ self.assertIn(archive, archives.select())
=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py 2010-08-01 04:34:26 +0000
+++ lib/lp/testing/factory.py 2010-08-02 00:41:06 +0000
@@ -144,7 +144,7 @@
from lp.soyuz.adapters.packagelocation import PackageLocation
from lp.soyuz.interfaces.archive import (
- default_name_by_purpose, IArchiveSet, ArchivePurpose)
+ default_name_by_purpose, IArchiveSet, ArchivePurpose, ArchiveStatus)
from lp.soyuz.interfaces.binarypackagebuild import IBinaryPackageBuildSet
from lp.soyuz.interfaces.binarypackagerelease import BinaryPackageFormat
from lp.soyuz.interfaces.component import IComponentSet
@@ -1731,7 +1731,8 @@
def makeArchive(self, distribution=None, owner=None, name=None,
purpose=None, enabled=True, private=False,
- virtualized=True, description=None, displayname=None):
+ virtualized=True, description=None, displayname=None,
+ commercial=False, status=None):
"""Create and return a new arbitrary archive.
:param distribution: Supply IDistribution, defaults to a new one
@@ -1744,6 +1745,8 @@
:param private: Whether the archive is created private.
:param virtualized: Whether the archive is virtualized.
:param description: A description of the archive.
+ :param commercial: Whether the archive is commercial.
+ :param status: The ArchiveStatus of the archive.
"""
if purpose is None:
purpose = ArchivePurpose.PPA
@@ -1772,11 +1775,17 @@
enabled=enabled, require_virtualized=virtualized,
description=description)
- if private:
+ if private or commercial:
naked_archive = removeSecurityProxy(archive)
naked_archive.private = True
naked_archive.buildd_secret = "sekrit"
+ if commercial:
+ removeSecurityProxy(archive).commercial = True
+
+ if status is not None:
+ removeSecurityProxy(archive).status = status
+
return archive
def makeBuilder(self, processor=None, url=None, name=None, title=None,
Follow ups
