launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #12440
[Merge] lp:~wallyworld/launchpad/distros-shared-with-person-1055940 into lp:launchpad
Ian Booth has proposed merging lp:~wallyworld/launchpad/distros-shared-with-person-1055940 into lp:launchpad with lp:~wallyworld/launchpad/projects-shared-with-person-1053387 as a prerequisite.
Commit message:
Add API to sharing service for finding distros shared with a user.
Requested reviews:
Curtis Hovey (sinzui)
Related bugs:
Bug #1055940 in Launchpad itself: "Cannot get the distributions that are shared with a person"
https://bugs.launchpad.net/launchpad/+bug/1055940
For more details, see:
https://code.launchpad.net/~wallyworld/launchpad/distros-shared-with-person-1055940/+merge/126164
== Implementation ==
Refactor the common logic for finding the shared products/distros into a helper method and add the necessary boiler plate to make it all hang together.
== Tests ==
Add some tests to test_sharingservice similar to the previously added tests for finding shared products. Make a few tweaks so that code can be resued.
== Lint ==
Checking for conflicts and issues in changed files.
Linting changed files:
lib/lp/registry/interfaces/sharingservice.py
lib/lp/registry/services/sharingservice.py
lib/lp/registry/services/tests/test_sharingservice.py
--
https://code.launchpad.net/~wallyworld/launchpad/distros-shared-with-person-1055940/+merge/126164
Your team Launchpad code reviewers is subscribed to branch lp:launchpad.
=== modified file 'lib/lp/registry/interfaces/sharingservice.py'
--- lib/lp/registry/interfaces/sharingservice.py 2012-09-25 06:20:37 +0000
+++ lib/lp/registry/interfaces/sharingservice.py 2012-09-25 06:20:37 +0000
@@ -38,6 +38,7 @@
SharingPermission,
SpecificationSharingPolicy,
)
+from lp.registry.interfaces.distribution import IDistribution
from lp.registry.interfaces.person import IPerson
from lp.registry.interfaces.pillar import IPillar
from lp.registry.interfaces.product import IProduct
@@ -83,6 +84,22 @@
@export_read_operation()
@call_with(user=REQUEST_USER)
@operation_parameters(
+ person=Reference(IPerson, title=_('Person'), required=True))
+ @operation_returns_collection_of(IDistribution)
+ @operation_for_version('devel')
+ def getSharedDistributions(person, user):
+ """Find distributions for which person has one or more access policy
+ grants.
+
+ :param user: the user making the request. If the user is an admin, then
+ all distributions are returned, else only those for which the user
+ is a maintainer or driver.
+ :return: a collection of distributions
+ """
+
+ @export_read_operation()
+ @call_with(user=REQUEST_USER)
+ @operation_parameters(
pillar=Reference(IPillar, title=_('Pillar'), required=True),
person=Reference(IPerson, title=_('Person'), required=True))
@operation_for_version('devel')
=== modified file 'lib/lp/registry/services/sharingservice.py'
--- lib/lp/registry/services/sharingservice.py 2012-09-25 06:20:37 +0000
+++ lib/lp/registry/services/sharingservice.py 2012-09-25 06:20:37 +0000
@@ -66,6 +66,7 @@
AccessPolicyGrantFlat,
)
from lp.registry.model.commercialsubscription import CommercialSubscription
+from lp.registry.model.distribution import Distribution
from lp.registry.model.person import Person
from lp.registry.model.teammembership import TeamParticipation
from lp.registry.model.product import Product
@@ -125,8 +126,13 @@
AccessPolicy.id.is_in(ids)
)
- def getSharedProducts(self, person, user):
- """See `ISharingService`."""
+ def _getSharedPillars(self, person, user, pillar_class, extra_filter=None):
+ """Helper method for getSharedProducts and getSharedDistributions.
+
+ pillar_class is either Product or Distribution. Products define the
+ owner foreign key attribute as _owner so we need to account for that,
+ but otherwise the logic is the same for both pillar types.
+ """
if user is None:
return []
store = IStore(AccessPolicyGrantFlat)
@@ -134,35 +140,51 @@
if roles.in_admin:
filter = True
else:
- commercial_filter = None
- if roles.in_commercial_admin:
- commercial_filter = Exists(Select(
- 1, tables=CommercialSubscription,
- where=CommercialSubscription.product == Product.id))
with_statement = With("teams",
Select(TeamParticipation.teamID,
tables=TeamParticipation,
where=TeamParticipation.person == user.id))
teams_sql = SQL("SELECT team from teams")
store = store.with_(with_statement)
+ if IProduct.implementedBy(pillar_class):
+ ownerID = pillar_class._ownerID
+ else:
+ ownerID = pillar_class.ownerID
filter = Or(
- commercial_filter or False,
- Product._ownerID.is_in(teams_sql),
- Product.driverID.is_in(teams_sql))
+ extra_filter or False,
+ ownerID.is_in(teams_sql),
+ pillar_class.driverID.is_in(teams_sql))
tables = [
AccessPolicyGrantFlat,
Join(
AccessPolicy,
AccessPolicyGrantFlat.policy_id == AccessPolicy.id)]
+ if IProduct.implementedBy(pillar_class):
+ access_policy_column = AccessPolicy.product_id
+ else:
+ access_policy_column = AccessPolicy.distribution_id
result_set = store.find(
- Product,
- Product.id.is_in(
+ pillar_class,
+ pillar_class.id.is_in(
Select(
- columns=AccessPolicy.product_id, tables=tables,
+ columns=access_policy_column, tables=tables,
where=(AccessPolicyGrantFlat.grantee_id == person.id))
), filter)
return result_set
+ def getSharedProducts(self, person, user):
+ """See `ISharingService`."""
+ commercial_filter = None
+ if user and IPersonRoles(user).in_commercial_admin:
+ commercial_filter = Exists(Select(
+ 1, tables=CommercialSubscription,
+ where=CommercialSubscription.product == Product.id))
+ return self._getSharedPillars(person, user, Product, commercial_filter)
+
+ def getSharedDistributions(self, person, user):
+ """See `ISharingService`."""
+ return self._getSharedPillars(person, user, Distribution)
+
@available_with_permission('launchpad.Driver', 'pillar')
def getSharedArtifacts(self, pillar, person, user, include_bugs=True,
include_branches=True, include_specifications=True):
=== modified file 'lib/lp/registry/services/tests/test_sharingservice.py'
--- lib/lp/registry/services/tests/test_sharingservice.py 2012-09-25 06:20:37 +0000
+++ lib/lp/registry/services/tests/test_sharingservice.py 2012-09-25 06:20:37 +0000
@@ -1380,6 +1380,57 @@
product = self.factory.makeProduct(driver=driver_team)
self._assert_getSharedProducts(product, driver_team.teamowner)
+ def _assert_getSharedDistributions(self, distro, who=None):
+ # Test that 'who' can query the shared distros for a grantee.
+
+ # Make a distro not related to 'who' which will be shared.
+ unrelated_distro = self.factory.makeDistribution()
+ # Make an unshared distro.
+ self.factory.makeDistribution()
+ person = self.factory.makePerson()
+ # Include more than one permission to ensure distinct works.
+ permissions = {
+ InformationType.PRIVATESECURITY: SharingPermission.ALL,
+ InformationType.USERDATA: SharingPermission.ALL}
+ with person_logged_in(distro.owner):
+ self.service.sharePillarInformation(
+ distro, person, distro.owner, permissions)
+ with person_logged_in(unrelated_distro.owner):
+ self.service.sharePillarInformation(
+ unrelated_distro, person, unrelated_distro.owner,
+ permissions)
+ shared = self.service.getSharedDistributions(person, who)
+ expected = []
+ if who:
+ expected = [distro]
+ if IPersonRoles(who).in_admin:
+ expected.append(unrelated_distro)
+ self.assertEqual(expected, list(shared))
+
+ def test_getSharedDistributions_anonymous(self):
+ # Anonymous users don't get to see any shared distros.
+ distro = self.factory.makeDistribution()
+ self._assert_getSharedDistributions(distro)
+
+ def test_getSharedDistributions_admin(self):
+ # Admins can see all shared distros.
+ admin = getUtility(ILaunchpadCelebrities).admin.teamowner
+ distro = self.factory.makeDistribution()
+ self._assert_getSharedDistributions(distro, admin)
+
+ def test_getSharedDistributions_owner(self):
+ # Users only see shared distros they own.
+ owner_team = self.factory.makeTeam(
+ membership_policy=TeamMembershipPolicy.MODERATED)
+ distro = self.factory.makeDistribution(owner=owner_team)
+ self._assert_getSharedDistributions(distro, owner_team.teamowner)
+
+ def test_getSharedDistributions_driver(self):
+ # Users only see shared distros they are the driver for.
+ driver_team = self.factory.makeTeam()
+ distro = self.factory.makeDistribution(driver=driver_team)
+ self._assert_getSharedDistributions(distro, driver_team.teamowner)
+
def test_getSharedBugs(self):
# Test the getSharedBugs method.
owner = self.factory.makePerson()
@@ -1720,7 +1771,7 @@
super(TestWebService, self).setUp()
self.webservice = LaunchpadWebServiceCaller(
'launchpad-library', 'salgado-change-anything')
- self._sharePillarInformation()
+ self._sharePillarInformation(self.pillar)
def test_url(self):
# Test that the url for the service is correct.
@@ -1745,8 +1796,8 @@
return self._named_get(
'getPillarGranteeData', pillar=pillar_uri)
- def _sharePillarInformation(self):
- pillar_uri = canonical_url(self.pillar, force_local_path=True)
+ def _sharePillarInformation(self, pillar):
+ pillar_uri = canonical_url(pillar, force_local_path=True)
return self._named_post(
'sharePillarInformation', pillar=pillar_uri,
grantee=self.grantee_uri,
@@ -1766,14 +1817,14 @@
self.launchpad = self.factory.makeLaunchpadService(person=self.owner)
self.service = self.launchpad.load('+services/sharing')
transaction.commit()
- self._sharePillarInformation()
+ self._sharePillarInformation(self.pillar)
def _getPillarGranteeData(self):
ws_pillar = ws_object(self.launchpad, self.pillar)
return self.service.getPillarGranteeData(pillar=ws_pillar)
- def _sharePillarInformation(self):
- ws_pillar = ws_object(self.launchpad, self.pillar)
+ def _sharePillarInformation(self, pillar):
+ ws_pillar = ws_object(self.launchpad, pillar)
ws_grantee = ws_object(self.launchpad, self.grantee)
return self.service.sharePillarInformation(pillar=ws_pillar,
grantee=ws_grantee,
@@ -1788,6 +1839,16 @@
self.assertEqual(1, len(products))
self.assertEqual(products[0].name, self.pillar.name)
+ def test_getSharedDistributions(self):
+ # Test the exported getSharedDistributions() method.
+ distro = self.factory.makeDistribution(owner=self.owner)
+ transaction.commit()
+ self._sharePillarInformation(distro)
+ ws_grantee = ws_object(self.launchpad, self.grantee)
+ distros = self.service.getSharedDistributions(person=ws_grantee)
+ self.assertEqual(1, len(distros))
+ self.assertEqual(distros[0].name, distro.name)
+
def test_getSharedBugs(self):
# Test the exported getSharedBugs() method.
ws_pillar = ws_object(self.launchpad, self.pillar)
Follow ups