Merge lp:~adeuring/launchpad/sharingjob-remove-bp-subscriptions into lp:launchpad

[Abel Deuring]

specifications added to RemoveArtifactSubscriptionsJob.

This branch extends RemoveArtifactSubscriptionsJob so that it
removes subscriptions to specifications if a subscriber does no longer
have grants to access these specifications.

The change in lib/lp/registry/model/sharingjob.py is mostly
a mechanical change that adds specification handling to bug and branch
handling.

The method issues a Storm query to find subscribers of a specification
without access grants. The Storm expression "person with[out] access
to a spec" is defined in a new function get_specification_privacy_filter(),
similar to the existing functions get_bug_privacy_filter() and
get_branch_privacy_filter().

RemoveArtifactSubscriptionsJob accepted not only bugs and branches but
arbitrary objects, which I found a bit confusing while working on this
branch, so I added a check to ensure that only those objects can be
passed to this object that the class can handle, i.e. bugs, branches
and blueprints.

Many tests of RemoveArtifactSubscriptionsJob rely on two methods
_assert_bug_change_unsubscribes() and _assert_branch_change_unsubscribes()
which were quite long but nearly identical. The new test
test_change_information_type_specification() requires a third
variant of these methods, _assert_specification_change_unsubscribes().

I refactored all three _assert.* methods to use a common method.

tests:

./bin/test -vvt lp.registry.tests.test_sharingjob
./bin/test -vvt lp.blueprints.tests.test_specification.*test_get_specification_privacy_filter

no lint

[Benji York (benji)]

This branch looks good. Here are some minor comments:

It seems you have an extra underscore in the test name
"test_create__bad_artifact_class".

I like the generalization of _assert_artifact_change_unsubscribes. It is
unfortunate that the function is so long though. I wonder if instead of
passing in a test type we could instead break out the contents of that if block
into functions that are instead passed in and called.

Another idea: break the _assert_artifact_change_unsubscribes into two
functions, one for the code before "if test_type" and one after. The
individual assertions could then call the first function, then include their
unique code (currently contained in the if) and then finally call the last
function.

review: Approve (code)

[Abel Deuring (adeuring)]

Benji, thanks for these suggestions. I chose the first variant, moved the "if test_type == " blocks into the callsites. _assert_artifact_change_unsubscribes() is still quite long, but at least a bit more readable...

[Benji York (benji)]

On Mon, Sep 24, 2012 at 2:16 PM, Abel Deuring
<email address hidden> wrote:
> Benji, thanks for these suggestions. I chose the first variant, moved the "if
> test_type == " blocks into the callsites.
> _assert_artifact_change_unsubscribes() is still quite long, but at least a
> bit more readable...

Cool.

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2012-09-21 00:08:41 +0000
+++ database/schema/security.cfg	2012-09-24 14:20:26 +0000
@@ -1966,6 +1966,8 @@
 public.person                           = SELECT
 public.product                          = SELECT
 public.sharingjob                       = SELECT, INSERT, UPDATE
+public.specification                    = SELECT
+public.specificationsubscription        = SELECT, DELETE
 public.teamparticipation                = SELECT
 type=user
 

=== modified file 'lib/lp/blueprints/model/specification.py'
--- lib/lp/blueprints/model/specification.py	2012-09-18 19:41:02 +0000
+++ lib/lp/blueprints/model/specification.py	2012-09-24 14:20:26 +0000
@@ -5,6 +5,7 @@
 
 __metaclass__ = type
 __all__ = [
+    'get_specification_privacy_filter',
     'HasSpecificationsMixin',
     'recursive_blocked_query',
     'recursive_dependent_query',
@@ -31,6 +32,7 @@
     And,
     In,
     Join,
+    LeftJoin,
     Or,
     Select,
     )
@@ -724,14 +726,15 @@
         notify(ObjectCreatedEvent(sub, user=subscribed_by))
         return sub
 
-    def unsubscribe(self, person, unsubscribed_by):
+    def unsubscribe(self, person, unsubscribed_by, ignore_permissions=False):
         """See ISpecification."""
         # see if a relevant subscription exists, and if so, delete it
         if person is None:
             person = unsubscribed_by
         for sub in self.subscriptions:
             if sub.person.id == person.id:
-                if not sub.canBeUnsubscribedByUser(unsubscribed_by):
+                if (not sub.canBeUnsubscribedByUser(unsubscribed_by) and
+                    not ignore_permissions):
                     raise UserCannotUnsubscribePerson(
                         '%s does not have permission to unsubscribe %s.' % (
                             unsubscribed_by.displayname,
@@ -1226,3 +1229,55 @@
     def get(self, spec_id):
         """See lp.blueprints.interfaces.specification.ISpecificationSet."""
         return Specification.get(spec_id)
+
+
+def get_specification_privacy_filter(user):
+    """Return a Storm expression for filtering specifications by privacy.
+
+    :param user: A Person ID or a column reference.
+    :return: A Storm expression to check if a peron has access grants
+         for a specification.
+    """
+    # Avoid circular imports.
+    from lp.registry.model.accesspolicy import (
+        AccessArtifact,
+        AccessPolicy,
+        AccessPolicyGrantFlat,
+        )
+    public_specification_filter = (
+        Specification.information_type.is_in(PUBLIC_INFORMATION_TYPES))
+    if user is None:
+        return public_specification_filter
+    return Or(
+        public_specification_filter,
+        Specification.id.is_in(
+            Select(
+                Specification.id,
+                tables=(
+                    Specification,
+                    Join(
+                        AccessPolicy,
+                        And(
+                            Or(
+                                Specification.productID ==
+                                    AccessPolicy.product_id,
+                                Specification.distributionID ==
+                                    AccessPolicy.distribution_id),
+                            Specification.information_type ==
+                                AccessPolicy.type)),
+                    Join(
+                        AccessPolicyGrantFlat,
+                        AccessPolicy.id == AccessPolicyGrantFlat.policy_id),
+                    LeftJoin(
+                        AccessArtifact,
+                        AccessPolicyGrantFlat.abstract_artifact_id ==
+                            AccessArtifact.id),
+                    Join(
+                        TeamParticipation,
+                        And(
+                            TeamParticipation.team ==
+                                AccessPolicyGrantFlat.grantee_id,
+                            TeamParticipation.person == user))),
+                where=Or(
+                    AccessPolicyGrantFlat.abstract_artifact_id == None,
+                    AccessArtifact.specification_id == Specification.id))))

=== modified file 'lib/lp/blueprints/tests/test_specification.py'
--- lib/lp/blueprints/tests/test_specification.py	2012-09-17 15:19:10 +0000
+++ lib/lp/blueprints/tests/test_specification.py	2012-09-24 14:20:26 +0000
@@ -6,6 +6,7 @@
 __metaclass__ = type
 
 
+from storm.store import Store
 from zope.component import (
     getUtility,
     queryAdapter,
@@ -31,7 +32,14 @@
     )
 from lp.blueprints.errors import TargetAlreadyHasSpecification
 from lp.blueprints.interfaces.specification import ISpecificationSet
-from lp.registry.enums import SharingPermission
+from lp.blueprints.model.specification import (
+    get_specification_privacy_filter,
+    Specification,
+    )
+from lp.registry.enums import (
+    SharingPermission,
+    SpecificationSharingPolicy,
+    )
 from lp.security import (
     AdminSpecification,
     EditSpecificationByRelatedPeople,
@@ -409,6 +417,49 @@
                 specification.target.owner, specification,
                 error_expected=False, attribute='name', value='foo')
 
+    def test_get_specification_privacy_filter(self):
+        # get_specification_privacy_filter() returns a Storm expression
+        # that can be used to filter specifications by their visibility-
+        owner = self.factory.makePerson()
+        product = self.factory.makeProduct(
+            owner=owner,
+            specification_sharing_policy=(
+                SpecificationSharingPolicy.PUBLIC_OR_PROPRIETARY))
+        public_spec = self.factory.makeSpecification(product=product)
+        proprietary_spec_1 = self.factory.makeSpecification(
+            product=product, information_type=InformationType.PROPRIETARY)
+        proprietary_spec_2 = self.factory.makeSpecification(
+            product=product, information_type=InformationType.PROPRIETARY)
+        all_specs = [
+            public_spec, proprietary_spec_1, proprietary_spec_2]
+        store = Store.of(product)
+        specs_for_anon = store.find(
+            Specification, get_specification_privacy_filter(None),
+            Specification.productID == product.id)
+        self.assertContentEqual([public_spec], specs_for_anon)
+        # Product owners havae grants on the product, the privacy
+        # filter returns thus all specifications for them.
+        specs_for_owner = store.find(
+            Specification, get_specification_privacy_filter(owner.id),
+            Specification.productID == product.id)
+        self.assertContentEqual(all_specs, specs_for_owner)
+        # The filter returns only public specs for ordinary users.
+        user = self.factory.makePerson()
+        specs_for_other_user = store.find(
+            Specification, get_specification_privacy_filter(user.id),
+            Specification.productID == product.id)
+        self.assertContentEqual([public_spec], specs_for_other_user)
+        # If the user has a grant for a specification, the filter returns
+        # this specification too.
+        with person_logged_in(owner):
+            getUtility(IService, 'sharing').ensureAccessGrants(
+                [user], owner, specifications=[proprietary_spec_1])
+        specs_for_other_user = store.find(
+            Specification, get_specification_privacy_filter(user.id),
+            Specification.productID == product.id)
+        self.assertContentEqual(
+            [public_spec, proprietary_spec_1], specs_for_other_user)
+
 
 class TestSpecificationSet(TestCaseWithFactory):
 

=== modified file 'lib/lp/registry/model/sharingjob.py'
--- lib/lp/registry/model/sharingjob.py	2012-09-18 19:41:02 +0000
+++ lib/lp/registry/model/sharingjob.py	2012-09-24 14:20:26 +0000
@@ -46,6 +46,14 @@
     IBug,
     IBugSet,
     )
+from lp.blueprints.interfaces.specification import ISpecification
+from lp.blueprints.model.specification import (
+    get_specification_privacy_filter,
+    Specification,
+    )
+from lp.blueprints.model.specificationsubscription import (
+    SpecificationSubscription,
+    )
 from lp.bugs.model.bugsubscription import BugSubscription
 from lp.bugs.model.bugtaskflat import BugTaskFlat
 from lp.bugs.model.bugtasksearch import get_bug_privacy_filter_terms
@@ -265,18 +273,25 @@
 
         bug_ids = []
         branch_ids = []
+        specification_ids = []
         if artifacts:
             for artifact in artifacts:
                 if IBug.providedBy(artifact):
                     bug_ids.append(artifact.id)
                 elif IBranch.providedBy(artifact):
                     branch_ids.append(artifact.id)
+                elif ISpecification.providedBy(artifact):
+                    specification_ids.append(artifact.id)
+                else:
+                    raise ValueError(
+                        'Unsupported artifact: %r' % artifact)
         information_types = [
             info_type.value for info_type in information_types or []
         ]
         metadata = {
             'bug_ids': bug_ids,
             'branch_ids': branch_ids,
+            'specification_ids': specification_ids,
             'information_types': information_types,
             'requestor.id': requestor.id
         }
@@ -308,6 +323,10 @@
         return [getUtility(IBranchLookup).get(id) for id in self.branch_ids]
 
     @property
+    def specification_ids(self):
+        return self.metadata.get('specification_ids', [])
+
+    @property
     def information_types(self):
         if not 'information_types' in self.metadata:
             return []
@@ -332,6 +351,7 @@
             'requestor': self.requestor.name,
             'bug_ids': self.bug_ids,
             'branch_ids': self.branch_ids,
+            'specification_ids': self.specification_ids,
             'pillar': getattr(self.pillar, 'name', None),
             'grantee': getattr(self.grantee, 'name', None)
             }
@@ -346,9 +366,13 @@
 
         bug_filters = []
         branch_filters = []
+        specification_filters = []
 
         if self.branch_ids:
             branch_filters.append(Branch.id.is_in(self.branch_ids))
+        if self.specification_ids:
+            specification_filters.append(Specification.id.is_in(
+                self.specification_ids))
         if self.bug_ids:
             bug_filters.append(BugTaskFlat.bug_id.is_in(self.bug_ids))
         else:
@@ -358,14 +382,21 @@
                         self.information_types))
                 branch_filters.append(
                     Branch.information_type.is_in(self.information_types))
+                specification_filters.append(
+                    Specification.information_type.is_in(
+                        self.information_types))
             if self.product:
                 bug_filters.append(
                     BugTaskFlat.product == self.product)
                 branch_filters.append(Branch.product == self.product)
+                specification_filters.append(
+                    Specification.product == self.product)
             if self.distro:
                 bug_filters.append(
                     BugTaskFlat.distribution == self.distro)
                 branch_filters.append(Branch.distribution == self.distro)
+                specification_filters.append(
+                    Specification.distribution == self.distro)
 
         if self.grantee:
             bug_filters.append(
@@ -378,6 +409,11 @@
                     Select(
                         TeamParticipation.personID,
                         where=TeamParticipation.team == self.grantee)))
+            specification_filters.append(
+                In(SpecificationSubscription.personID,
+                    Select(
+                        TeamParticipation.personID,
+                        where=TeamParticipation.team == self.grantee)))
 
         if bug_filters:
             bug_filters.append(Not(
@@ -402,3 +438,20 @@
             for sub in branch_subscriptions:
                 sub.branch.unsubscribe(
                     sub.person, self.requestor, ignore_permissions=True)
+        if specification_filters:
+            specification_filters.append(
+                Not(get_specification_privacy_filter(
+                    SpecificationSubscription.personID)))
+            tables = (
+                SpecificationSubscription,
+                Join(
+                    Specification,
+                    Specification.id ==
+                        SpecificationSubscription.specificationID))
+            specifications_subscriptions = IStore(
+                SpecificationSubscription).using(*tables).find(
+                SpecificationSubscription, *specification_filters).config(
+                distinct=True)
+            for sub in specifications_subscriptions:
+                sub.specification.unsubscribe(
+                    sub.person, self.requestor, ignore_permissions=True)

=== modified file 'lib/lp/registry/tests/test_sharingjob.py'
--- lib/lp/registry/tests/test_sharingjob.py	2012-09-18 19:41:02 +0000
+++ lib/lp/registry/tests/test_sharingjob.py	2012-09-24 14:20:26 +0000
@@ -17,6 +17,7 @@
     BranchSubscriptionNotificationLevel,
     CodeReviewNotificationLevel,
     )
+from lp.registry.enums import SpecificationSharingPolicy
 from lp.registry.interfaces.accesspolicy import (
     IAccessArtifactGrantSource,
     IAccessPolicySource,
@@ -117,6 +118,16 @@
             'for branch_ids=[%d], requestor=%s>'
             % (branch.id, requestor.name), repr(job))
 
+    def test_repr_specifications(self):
+        requestor = self.factory.makePerson()
+        specification = self.factory.makeSpecification()
+        job = getUtility(IRemoveArtifactSubscriptionsJobSource).create(
+            requestor, artifacts=[specification])
+        self.assertEqual(
+            '<REMOVE_ARTIFACT_SUBSCRIPTIONS job reconciling subscriptions '
+            'for requestor=%s, specification_ids=[%d]>'
+            % (requestor.name, specification.id), repr(job))
+
     def test_create_success(self):
         # Create an instance of SharingJobDerived that delegates to SharingJob.
         self.assertIs(True, ISharingJobSource.providedBy(SharingJobDerived))
@@ -244,6 +255,17 @@
         self.assertEqual(requestor.id, naked_job.requestor_id)
         self.assertContentEqual([bug.id], naked_job.bug_ids)
 
+    def test_create__bad_artifact_class(self):
+        # A ValueError is raised if an object of an unsupported type
+        # is passed as an artifact to
+        # IRemoveArtifactSubscriptionsJob.create().
+        requestor = self.factory.makePerson()
+        wrong_object = self.factory.makePerson()
+        self.assertRaises(
+            ValueError,
+            getUtility(IRemoveArtifactSubscriptionsJobSource).create,
+            requestor, [wrong_object])
+
     def test_getErrorRecipients(self):
         # The pillar owner and job requestor are the error recipients.
         requestor = self.factory.makePerson()
@@ -257,10 +279,15 @@
         self.assertContentEqual(
             expected_emails, job.getErrorRecipients())
 
-    def _assert_bug_change_unsubscribes(self, change_callback):
-        # Subscribers are unsubscribed if the bug becomes invisible due to a
-        # change in information_type.
-        product = self.factory.makeProduct()
+    def _assert_artifact_change_unsubscribes(self, change_callback,
+                                             configure_test):
+        # Subscribers are unsubscribed if the artifact becomes invisible
+        # due to a change in information_type.
+        product = self.factory.makeProduct(
+            specification_sharing_policy=(
+                SpecificationSharingPolicy.EMBARGOED_OR_PROPRIETARY))
+        self.factory.makeAccessPolicy(product, InformationType.PROPRIETARY)
+        self.factory.makeAccessPolicy(product, InformationType.EMBARGOED)
         owner = self.factory.makePerson()
         [policy] = getUtility(IAccessPolicySource).find(
             [(product, InformationType.USERDATA)])
@@ -278,6 +305,9 @@
         branch = self.factory.makeBranch(
             owner=owner, product=product,
             information_type=InformationType.USERDATA)
+        specification = self.factory.makeSpecification(
+            owner=owner, product=product,
+            information_type=InformationType.PROPRIETARY)
 
         # The artifact grantees will not lose access when the job is run.
         artifact_indirect_grantee = self.factory.makePerson()
@@ -285,113 +315,127 @@
             membership_policy=TeamMembershipPolicy.RESTRICTED,
             members=[artifact_indirect_grantee])
 
-        bug.subscribe(policy_team_grantee, owner)
-        bug.subscribe(policy_indirect_grantee, owner)
-        bug.subscribe(artifact_team_grantee, owner)
         bug.subscribe(artifact_indirect_grantee, owner)
         branch.subscribe(artifact_indirect_grantee,
             BranchSubscriptionNotificationLevel.NOEMAIL, None,
             CodeReviewNotificationLevel.NOEMAIL, owner)
+        # Subscribing somebody to a specification does not automatically
+        # create an artifact grant.
+        spec_artifact = self.factory.makeAccessArtifact(specification)
+        self.factory.makeAccessArtifactGrant(
+            spec_artifact, artifact_indirect_grantee)
+        specification.subscribe(artifact_indirect_grantee, owner)
+
+        # pick one of the concrete artifacts (bug, branch or spec)
+        # and subscribe the teams and persons.
+        concrete_artifact, get_pillars, get_subscribers = configure_test(
+            bug, branch, specification, policy_team_grantee,
+            policy_indirect_grantee, artifact_team_grantee, owner)
+
         # Subscribing policy_team_grantee has created an artifact grant so we
         # need to revoke that to test the job.
-        artifact = self.factory.makeAccessArtifact(concrete=bug)
+        artifact = self.factory.makeAccessArtifact(concrete=concrete_artifact)
         getUtility(IAccessArtifactGrantSource).revokeByArtifact(
             [artifact], [policy_team_grantee])
 
         # policy grantees are subscribed because the job has not been run yet.
-        subscribers = removeSecurityProxy(bug).getDirectSubscribers()
+        subscribers = get_subscribers(concrete_artifact)
         self.assertIn(policy_team_grantee, subscribers)
         self.assertIn(policy_indirect_grantee, subscribers)
 
-        # Change bug attributes so that it can become inaccessible for
+        # Change artifact attributes so that it can become inaccessible for
         # some users.
-        change_callback(bug)
+        change_callback(concrete_artifact)
         reconcile_access_for_artifact(
-            bug, bug.information_type, bug.affected_pillars)
+            concrete_artifact, concrete_artifact.information_type,
+            get_pillars(concrete_artifact))
 
         getUtility(IRemoveArtifactSubscriptionsJobSource).create(
-            owner, [bug])
+            owner, [concrete_artifact])
         with block_on_job(self):
             transaction.commit()
 
         # Check the result. Policy grantees will be unsubscribed.
-        subscribers = removeSecurityProxy(bug).getDirectSubscribers()
+        subscribers = get_subscribers(concrete_artifact)
         self.assertNotIn(policy_team_grantee, subscribers)
         self.assertNotIn(policy_indirect_grantee, subscribers)
         self.assertIn(artifact_team_grantee, subscribers)
-        self.assertIn(artifact_indirect_grantee, subscribers)
+        self.assertIn(artifact_indirect_grantee, bug.getDirectSubscribers())
         self.assertIn(artifact_indirect_grantee, branch.subscribers)
+        self.assertIn(artifact_indirect_grantee, specification.subscribers)
+
+    def _assert_bug_change_unsubscribes(self, change_callback):
+
+        def get_pillars(concrete_artifact):
+            return concrete_artifact.affected_pillars
+
+        def get_subscribers(concrete_artifact):
+            return removeSecurityProxy(
+                concrete_artifact).getDirectSubscribers()
+
+        def configure_test(bug, branch, specification, policy_team_grantee,
+                           policy_indirect_grantee, artifact_team_grantee,
+                           owner):
+            concrete_artifact = bug
+            bug.subscribe(policy_team_grantee, owner)
+            bug.subscribe(policy_indirect_grantee, owner)
+            bug.subscribe(artifact_team_grantee, owner)
+            return concrete_artifact, get_pillars, get_subscribers
+
+        self._assert_artifact_change_unsubscribes(
+            change_callback, configure_test)
 
     def _assert_branch_change_unsubscribes(self, change_callback):
-        product = self.factory.makeProduct()
-        owner = self.factory.makePerson()
-        [policy] = getUtility(IAccessPolicySource).find(
-            [(product, InformationType.USERDATA)])
-        # The policy grantees will lose access.
-        policy_indirect_grantee = self.factory.makePerson()
-        policy_team_grantee = self.factory.makeTeam(
-            membership_policy=TeamMembershipPolicy.RESTRICTED,
-            members=[policy_indirect_grantee])
-
-        self.factory.makeAccessPolicyGrant(policy, policy_team_grantee, owner)
-        login_person(owner)
-        bug = self.factory.makeBug(
-            owner=owner, target=product,
-            information_type=InformationType.USERDATA)
-        branch = self.factory.makeBranch(
-            owner=owner, product=product,
-            information_type=InformationType.USERDATA)
-
-        # The artifact grantees will not lose access when the job is run.
-        artifact_indirect_grantee = self.factory.makePerson()
-        artifact_team_grantee = self.factory.makeTeam(
-            membership_policy=TeamMembershipPolicy.RESTRICTED,
-            members=[artifact_indirect_grantee])
-
-        branch.subscribe(
-            policy_team_grantee, BranchSubscriptionNotificationLevel.NOEMAIL,
-            None, CodeReviewNotificationLevel.NOEMAIL, owner)
-        branch.subscribe(
-            policy_indirect_grantee,
-            BranchSubscriptionNotificationLevel.NOEMAIL, None,
-            CodeReviewNotificationLevel.NOEMAIL, owner)
-        branch.subscribe(
-            artifact_team_grantee,
-            BranchSubscriptionNotificationLevel.NOEMAIL, None,
-            CodeReviewNotificationLevel.NOEMAIL, owner)
-        branch.subscribe(
-            artifact_indirect_grantee,
-            BranchSubscriptionNotificationLevel.NOEMAIL, None,
-            CodeReviewNotificationLevel.NOEMAIL, owner)
-        bug.subscribe(artifact_indirect_grantee, owner)
-        # Subscribing policy_team_grantee has created an artifact grant so we
-        # need to revoke that to test the job.
-        artifact = self.factory.makeAccessArtifact(concrete=branch)
-        getUtility(IAccessArtifactGrantSource).revokeByArtifact(
-            [artifact], [policy_team_grantee])
-
-        # policy grantees are subscribed because the job has not been run yet.
-        #subscribers = removeSecurityProxy(branch).subscribers
-        self.assertIn(policy_team_grantee, branch.subscribers)
-        self.assertIn(policy_indirect_grantee, branch.subscribers)
-
-        # Change branch attributes so that it can become inaccessible for
-        # some users.
-        change_callback(branch)
-        reconcile_access_for_artifact(
-            branch, branch.information_type, [branch.product])
-
-        getUtility(IRemoveArtifactSubscriptionsJobSource).create(
-            owner, [branch])
-        with block_on_job(self):
-            transaction.commit()
-
-        # Check the result. Policy grantees will be unsubscribed.
-        self.assertNotIn(policy_team_grantee, branch.subscribers)
-        self.assertNotIn(policy_indirect_grantee, branch.subscribers)
-        self.assertIn(artifact_team_grantee, branch.subscribers)
-        self.assertIn(artifact_indirect_grantee, branch.subscribers)
-        self.assertIn(artifact_indirect_grantee, bug.getDirectSubscribers())
+
+        def get_pillars(concrete_artifact):
+            return [concrete_artifact.product]
+
+        def get_subscribers(concrete_artifact):
+            return concrete_artifact.subscribers
+
+        def configure_test(bug, branch, specification, policy_team_grantee,
+                           policy_indirect_grantee, artifact_team_grantee,
+                           owner):
+            concrete_artifact = branch
+            branch.subscribe(
+                policy_team_grantee,
+                BranchSubscriptionNotificationLevel.NOEMAIL,
+                None, CodeReviewNotificationLevel.NOEMAIL, owner)
+            branch.subscribe(
+                policy_indirect_grantee,
+                BranchSubscriptionNotificationLevel.NOEMAIL, None,
+                CodeReviewNotificationLevel.NOEMAIL, owner)
+            branch.subscribe(
+                artifact_team_grantee,
+                BranchSubscriptionNotificationLevel.NOEMAIL, None,
+                CodeReviewNotificationLevel.NOEMAIL, owner)
+            return concrete_artifact, get_pillars, get_subscribers
+
+        self._assert_artifact_change_unsubscribes(
+            change_callback, configure_test)
+
+    def _assert_specification_change_unsubscribes(self, change_callback):
+
+        def get_pillars(concrete_artifact):
+            return [concrete_artifact.product]
+
+        def get_subscribers(concrete_artifact):
+            return concrete_artifact.subscribers
+
+        def configure_test(bug, branch, specification, policy_team_grantee,
+                           policy_indirect_grantee, artifact_team_grantee,
+                           owner):
+            concrete_artifact = specification
+            specification.subscribe(policy_team_grantee, owner)
+            specification.subscribe(policy_indirect_grantee, owner)
+            spec_artifact = self.factory.makeAccessArtifact(specification)
+            self.factory.makeAccessArtifactGrant(
+                spec_artifact, artifact_team_grantee)
+            specification.subscribe(artifact_team_grantee, owner)
+            return concrete_artifact, get_pillars, get_subscribers
+
+        self._assert_artifact_change_unsubscribes(
+            change_callback, configure_test)
 
     def test_change_information_type_branch(self):
         def change_information_type(branch):
@@ -400,6 +444,13 @@
 
         self._assert_branch_change_unsubscribes(change_information_type)
 
+    def test_change_information_type_specification(self):
+        def change_information_type(specification):
+            removeSecurityProxy(specification).information_type = (
+                InformationType.EMBARGOED)
+
+        self._assert_specification_change_unsubscribes(change_information_type)
+
     def test_change_information_type(self):
         # Changing the information type of a bug unsubscribes users who can no
         # longer see the bug.