launchpad-reviewers team mailing list archive

[Colin Watson <cjwatson@xxxxxxxxxxxxx>]

Review: Approve



Diff comments:

> diff --git a/lib/lp/registry/tests/test_personmerge.py b/lib/lp/registry/tests/test_personmerge.py
> index 02fefcc..4c55cb2 100644
> --- a/lib/lp/registry/tests/test_personmerge.py
> +++ b/lib/lp/registry/tests/test_personmerge.py
> @@ -664,6 +669,35 @@ class TestMergePeople(TestCaseWithFactory, KarmaTestMixin):
>          self.assertIsNone(snaps[1].git_path)
>          self.assertEqual(u'foo-1', snaps[1].name)
>  
> +    def test_merge_snapsubscription(self):
> +        # Checks that merging users moves subscriptions.
> +        self.useFixture(FeatureFixture(SNAP_TESTING_FLAGS))
> +        duplicate = self.factory.makePerson()
> +        mergee = self.factory.makePerson()
> +        snap = removeSecurityProxy(self.factory.makeSnap(
> +            owner=duplicate, registrant=duplicate,
> +            name=u'foo', private=True))
> +        self._do_premerge(duplicate, mergee)
> +        login_admin()

These tests normally call `_do_merge` after `login_person(mergee)`.  If you need to be an admin for some test setup before that, fair enough, but at least switch to the mergee for the actual merge.

> +        # Owner should have being subscribed automatically on creation.
> +        self.assertTrue(snap.visibleByUser(duplicate))
> +        self.assertThat(snap._getSubscription(duplicate), MatchesStructure(
> +            snap=Equals(snap),
> +            person=Equals(duplicate)
> +        ))
> +        self.assertFalse(snap.visibleByUser(mergee))
> +        self.assertIsNone(snap._getSubscription(mergee))
> +
> +        duplicate, mergee = self._do_merge(duplicate, mergee)
> +
> +        self.assertTrue(snap.visibleByUser(mergee))
> +        self.assertThat(snap._getSubscription(mergee), MatchesStructure(
> +            snap=Equals(snap),
> +            person=Equals(mergee)
> +        ))
> +        self.assertFalse(snap.visibleByUser(duplicate))
> +        self.assertIsNone(snap._getSubscription(duplicate))
> +
>      def test_merge_moves_oci_recipes(self):
>          # When person/teams are merged, oci recipes owned by the from
>          # person are moved.
> diff --git a/lib/lp/snappy/model/snap.py b/lib/lp/snappy/model/snap.py
> index 6b9af0a..1b98ad5 100644
> --- a/lib/lp/snappy/model/snap.py
> +++ b/lib/lp/snappy/model/snap.py
> @@ -1129,25 +1137,67 @@ class Snap(Storm, WebhookTargetMixin):
>  
>      def visibleByUser(self, user):
>          """See `ISnap`."""
> +        if self.information_type in PUBLIC_INFORMATION_TYPES:
> +            return True
> +        if user is None:
> +            return False
>          store = IStore(self)
>          return not store.find(
>              Snap,
>              Snap.id == self.id,
>              get_snap_privacy_filter(user)).is_empty()
>  
> +    def _getSubscription(self, person):
> +        """Returns person's subscription to this snap recipe, or None if no
> +        subscription is available.
> +        """
> +        if person is None:
> +            return None
> +        return Store.of(self).find(
> +            SnapSubscription,
> +            SnapSubscription.person == person,
> +            SnapSubscription.snap == self).one()
> +
> +    def _userCanBeSubscribed(self, person):
> +        """Checks if the given person can subscribe to this snap recipe."""
> +        return not (
> +            self.private and
> +            person.is_team and
> +            person.anyone_can_join())
> +
>      def subscribe(self, person, subscribed_by, ignore_permissions=False):
>          """See `ISnap`."""
> -        # XXX pappacena 2021-02-05: We may need a "SnapSubscription" here.
> +        if not self._userCanBeSubscribed(person):
> +            raise SubscriptionPrivacyViolation(
> +                "Open and delegated teams cannot be subscribed to private "
> +                "snap recipes.")
> +        subscription = self._getSubscription(person)
> +        if subscription is None:
> +            subscription = SnapSubscription(
> +                person=person, snap=self, subscribed_by=subscribed_by)
> +            Store.of(subscription).flush()
>          service = getUtility(IService, "sharing")
>          service.ensureAccessGrants(
>              [person], subscribed_by, snaps=[self],
>              ignore_permissions=ignore_permissions)
>  
> -    def unsubscribe(self, person, unsubscribed_by):
> +    def unsubscribe(self, person, unsubscribed_by, ignore_permissions=False):
>          """See `ISnap`."""
> -        service = getUtility(IService, "sharing")
> -        service.revokeAccessGrants(
> -            self.pillar, person, unsubscribed_by, snaps=[self])
> +        subscription = self._getSubscription(person)

Maybe silently return here if `subscription is None` (i.e. the user is already unsubscribed)?  We seem to do that elsewhere.

> +        if (not ignore_permissions
> +                and not subscription.canBeUnsubscribedByUser(unsubscribed_by)):
> +            raise UserCannotUnsubscribePerson(
> +                '%s does not have permission to unsubscribe %s.' % (
> +                    unsubscribed_by.displayname,
> +                    person.displayname))
> +        artifact = getUtility(IAccessArtifactSource).find([self])
> +        getUtility(IAccessArtifactGrantSource).revokeByArtifact(
> +            artifact, [person])
> +        # It should never be None, since we always create a SnapSubscription
> +        # on Snap.subscribe. But just in case...

I'm not sure how this follows; Snap.subscribe might not have been called for this particular person.

> +        if subscription is not None:
> +            store = Store.of(subscription)
> +            store.remove(subscription)
>          IStore(self).flush()
>  
>      def _reconcileAccess(self):
> diff --git a/lib/lp/snappy/model/snapsubscription.py b/lib/lp/snappy/model/snapsubscription.py
> new file mode 100644
> index 0000000..95a0026
> --- /dev/null
> +++ b/lib/lp/snappy/model/snapsubscription.py
> @@ -0,0 +1,61 @@
> +# Copyright 2020-2021 Canonical Ltd.  This software is licensed under the
> +# GNU Affero General Public License version 3 (see the file LICENSE).
> +
> +"""Snap subscription model."""
> +
> +from __future__ import absolute_import, print_function, unicode_literals
> +
> +__metaclass__ = type
> +__all__ = [
> +    'SnapSubscription'
> +]
> +
> +import pytz
> +from storm.properties import (
> +    DateTime,
> +    Int,
> +    )
> +from storm.references import Reference
> +from zope.interface import implementer
> +
> +from lp.registry.interfaces.person import validate_person
> +from lp.registry.interfaces.role import IPersonRoles
> +from lp.services.database.constants import UTC_NOW
> +from lp.services.database.stormbase import StormBase
> +from lp.snappy.interfaces.snapsubscription import ISnapSubscription
> +
> +
> +@implementer(ISnapSubscription)
> +class SnapSubscription(StormBase):
> +    """A relationship between a person and a snap recipe."""
> +
> +    __storm_table__ = 'SnapSubscription'
> +
> +    id = Int(primary=True)
> +
> +    person_id = Int(
> +        "person", allow_none=False, validator=validate_person)
> +    person = Reference(person_id, "Person.id")
> +
> +    snap_id = Int("snap", allow_none=False)
> +    snap = Reference(snap_id, "Snap.id")
> +
> +    date_created = DateTime(allow_none=False, default=UTC_NOW, tzinfo=pytz.UTC)
> +
> +    subscribed_by_id = Int(
> +        "subscribed_by", allow_none=False, validator=validate_person)
> +    subscribed_by = Reference(subscribed_by_id, "Person.id")
> +
> +    def __init__(self, snap=None, person=None, subscribed_by=None):

I'd be inclined to drop `=None` from all of these, since they're all required and have no reasonable defaults.

> +        super(SnapSubscription, self).__init__()
> +        self.snap = snap
> +        self.person = person
> +        self.subscribed_by = subscribed_by
> +
> +    def canBeUnsubscribedByUser(self, user):
> +        """See `ISnapSubscription`."""
> +        if user is None:
> +            return False
> +        return (user.inTeam(self.person) or
> +                user.inTeam(self.subscribed_by) or
> +                IPersonRoles(user).in_admin)

Shouldn't the snap owner be able to unsubscribe people too?

> diff --git a/lib/lp/snappy/tests/test_snap.py b/lib/lp/snappy/tests/test_snap.py
> index e8ea61e..bfa4e0c 100644
> --- a/lib/lp/snappy/tests/test_snap.py
> +++ b/lib/lp/snappy/tests/test_snap.py
> @@ -1353,6 +1356,9 @@ class TestSnapVisibility(TestCaseWithFactory):
>              AccessArtifactGrant.abstract_artifact_id == AccessArtifact.id,
>              *conditions)
>  
> +    def getSnapSubscription(self, snap, person):
> +        return removeSecurityProxy(snap)._getSubscription(person)

I wonder why `_getSubscription` couldn't just have its underscore dropped and be added to the interface, so that removing the security proxy isn't necessary.

> +
>      def test_only_owner_can_grant_access(self):
>          owner = self.factory.makePerson()
>          snap = self.factory.makeSnap(


-- 
https://code.launchpad.net/~pappacena/launchpad/+git/launchpad/+merge/397755
Your team Launchpad code reviewers is subscribed to branch ~pappacena/launchpad:snap-pillar-reconcile-access.