launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #05445
lp:~wallyworld/launchpad/pillar-owners-cannot-become-open-879103 into lp:launchpad
Ian Booth has proposed merging lp:~wallyworld/launchpad/pillar-owners-cannot-become-open-879103 into lp:launchpad with lp:~wallyworld/launchpad/no-open-pillar-owners-879103 as a prerequisite.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~wallyworld/launchpad/pillar-owners-cannot-become-open-879103/+merge/81532
This mp continues the work in the previous branch to prevent open teams from being pillar owners or security contacts. It prevents teams in these roles being edited to have an illegal policy. There are a lot of deletions so the diff looks bigger than it really is.
The next branch will fix the Add Member option to prevent illegal open/closed teams being added as members.
== Pre-Implementation ==
Talked to stub about query used to check for pillar ownership to ensure it performs.
== Implementation ==
There was an existing process used to prevent teams from having incorrect subscription policies:
- display all choices (valid or not)
- on form submission, invoke team_subscription_policy_can_transition() to validate the transition
- if there's a problem, display an error on the form
I don't like this approach - the user should not be given the choice to select invalid options. So I've changed how it's done:
1. Introduce 2 new enums - ClosedTeamSubscriptionPolicy and OpenTeamSubscriptionPolicy, each using the relevant closed/open items from TeamSubscriptionPolicy
2. Provide 2 new methods on ITeam:
- subscriptionPolicyMustBeClosed()
- subscriptionPolicyMustBeOpen()
The above use the existing business rules from team_subscription_policy_can_transition() plus new ones done for this mp.
3. In the setup for TeamEditView, invoke the methods in #2 above and set the subscriptionpolicy widget's vocab to be one of the new enums from #1 above as required.
4. So now the user only sees what they are allow to choose for the subscription policy.
This means that the form field validation for subscriptionpolicy can be deleted entirely:
- delete team_subscription_policy_can_transition()
- delete class TeamSubsciptionPolicyChoice(Choice)
IPersonRoles had 2 new methods added to provide the checks required to prevent open teams being pillar owners or security contacts:
- isPillarOwner()
- isSecurityContact()
The above are called from subscriptionPolicyMustBeClosed()
To prevent the api or other direct model manipulation from setting an illegal subscriptionpolicy, I've introduced a storm validator: validate_subscription_policy()
== Demo and QA ==
Screenshots of edit team pages showing subscription policy widget with only a subset of allowable values:
http://people.canonical.com/~ianb/onlyclosedteams.png
http://people.canonical.com/~ianb/onlyopenteams.png
== Tests ==
The existing tests for team_subscription_policy_can_transition() were re-purposed to test the new subscriptionPolicyMustBeClosed() and subscriptionPolicyMustBeOpen() methods. The test case is now called TestTeamSubscriptionPolicy:
- test_team_with_closed_super_team_cannot_be_open()
- test_team_with_open_super_team_can_be_open()
- test_team_with_active_ppas_cannot_be_open()
- test_team_without_active_ppas_can_be_open()
- test_team_with_private_bugs_cannot_be_open()
- test_team_with_private_bugs_assigned_cannot_be_open()
- test_team_with_open_sub_team_cannot_be_closed()
There are also tests for the validator:
- test_illegal_transition_to_open_subscription()
- test_illegal_transition_to_closed_subscription()
Tests added for the new IPersonRoles methods.
Add to TestPersonRoles:
- test_product_isOwner()
- test_projectgroup_isOwner()
- test_distribution_isOwner()
- test_product_isSecurityContact()
- test_distribution_isSecurityContact
Tests added to ensure the TestTeamEditView displays the correct subscriptionpolicy choices:
- test_edit_team_view_pillar_owner()
- test_edit_team_view_pillar_security_contact()
- test_edit_team_view_has_ppas()
- test_edit_team_view_has_closed_super_team()
- test_edit_team_view_subscribed_private_bug()
- test_edit_team_view_has_open_member()
== Lint ==
Checking for conflicts and issues in changed files.
Linting changed files:
lib/lp/registry/errors.py
lib/lp/registry/browser/team.py
lib/lp/registry/browser/tests/test_team_view.py
lib/lp/registry/interfaces/person.py
lib/lp/registry/interfaces/role.py
lib/lp/registry/model/person.py
lib/lp/registry/model/personroles.py
lib/lp/registry/tests/test_errors.py
lib/lp/registry/tests/test_personroles.py
lib/lp/registry/tests/test_team.py
--
https://code.launchpad.net/~wallyworld/launchpad/pillar-owners-cannot-become-open-879103/+merge/81532
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~wallyworld/launchpad/pillar-owners-cannot-become-open-879103 into lp:launchpad.
=== modified file 'lib/lp/registry/browser/team.py'
--- lib/lp/registry/browser/team.py 2011-10-06 00:16:41 +0000
+++ lib/lp/registry/browser/team.py 2011-11-08 10:47:33 +0000
@@ -136,12 +136,14 @@
MailingListAutoSubscribePolicy,
)
from lp.registry.interfaces.person import (
+ ClosedTeamSubscriptionPolicy,
ImmutableVisibilityError,
IPersonSet,
ITeam,
ITeamReassignment,
ITeamContactAddressForm,
ITeamCreation,
+ OpenTeamSubscriptionPolicy,
PersonVisibility,
PRIVATE_TEAM_PREFIX,
TeamContactMethod,
@@ -292,6 +294,16 @@
super(TeamEditView, self).setUpFields()
self.conditionallyOmitVisibility()
+ def setUpWidgets(self):
+ super(TeamEditView, self).setUpWidgets()
+ team = self.context
+ if team.subscriptionPolicyMustBeClosed():
+ self.widgets['subscriptionpolicy'].vocabulary = (
+ ClosedTeamSubscriptionPolicy)
+ if team.subscriptionPolicyMustBeOpen():
+ self.widgets['subscriptionpolicy'].vocabulary = (
+ OpenTeamSubscriptionPolicy)
+
@action('Save', name='save')
def action_save(self, action, data):
try:
=== modified file 'lib/lp/registry/browser/tests/test_team_view.py'
--- lib/lp/registry/browser/tests/test_team_view.py 2011-10-04 04:26:32 +0000
+++ lib/lp/registry/browser/tests/test_team_view.py 2011-11-08 10:47:33 +0000
@@ -18,9 +18,12 @@
)
from lp.registry.interfaces.mailinglist import MailingListStatus
from lp.registry.interfaces.person import (
+ ClosedTeamSubscriptionPolicy,
+ OpenTeamSubscriptionPolicy,
PersonVisibility,
+ TeamMembershipRenewalPolicy,
TeamSubscriptionPolicy,
- TeamMembershipRenewalPolicy)
+ )
from lp.soyuz.enums import ArchiveStatus
from lp.testing import (
login_person,
@@ -260,10 +263,99 @@
TeamSubscriptionPolicy.MODERATED,
view.widgets['subscriptionpolicy']._data)
self.assertEqual(
+ TeamSubscriptionPolicy,
+ view.widgets['subscriptionpolicy'].vocabulary)
+ self.assertEqual(
TeamMembershipRenewalPolicy.NONE,
view.widgets['renewal_policy']._data)
self.assertIsNone(view.widgets['defaultrenewalperiod']._data)
+ def _test_edit_team_view_expected_subscription_vocab(self,
+ fn_setup,
+ expected_vocab):
+ # The edit view renders only the specified policy choices when
+ # the setup performed by fn_setup occurs.
+ owner = self.factory.makePerson()
+ team = self.factory.makeTeam(
+ owner=owner, subscription_policy=TeamSubscriptionPolicy.MODERATED)
+ fn_setup(team)
+ with person_logged_in(owner):
+ view = create_initialized_view(team, name="+edit")
+ self.assertEqual(
+ expected_vocab, view.widgets['subscriptionpolicy'].vocabulary)
+
+ def test_edit_team_view_pillar_owner(self):
+ # The edit view renders only closed subscription policy choices when
+ # the team is a pillar owner.
+
+ def setup_team(team):
+ self.factory.makeProduct(owner=team)
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, ClosedTeamSubscriptionPolicy)
+
+ def test_edit_team_view_pillar_security_contact(self):
+ # The edit view renders only closed subscription policy choices when
+ # the team is a pillar security contact.
+
+ def setup_team(team):
+ self.factory.makeProduct(security_contact=team)
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, ClosedTeamSubscriptionPolicy)
+
+ def test_edit_team_view_has_ppas(self):
+ # The edit view renders only closed subscription policy choices when
+ # the team has any ppas.
+
+ def setup_team(team):
+ team.createPPA()
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, ClosedTeamSubscriptionPolicy)
+
+ def test_edit_team_view_has_closed_super_team(self):
+ # The edit view renders only closed subscription policy choices when
+ # the team has any closed super teams.
+
+ def setup_team(team):
+ super_team = self.factory.makeTeam(
+ owner=team.teamowner,
+ subscription_policy=TeamSubscriptionPolicy.RESTRICTED)
+ with person_logged_in(team.teamowner):
+ super_team.addMember(
+ team, team.teamowner, force_team_add=True)
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, ClosedTeamSubscriptionPolicy)
+
+ def test_edit_team_view_subscribed_private_bug(self):
+ # The edit view renders only closed subscription policy choices when
+ # the team is subscribed to a private bug.
+
+ def setup_team(team):
+ bug = self.factory.makeBug(owner=team.teamowner, private=True)
+ with person_logged_in(team.teamowner):
+ bug.default_bugtask.transitionToAssignee(team)
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, ClosedTeamSubscriptionPolicy)
+
+ def test_edit_team_view_has_open_member(self):
+ # The edit view renders open closed subscription policy choices when
+ # the team has any open sub teams.
+
+ def setup_team(team):
+ team_member = self.factory.makeTeam(
+ owner=team.teamowner,
+ subscription_policy=TeamSubscriptionPolicy.DELEGATED)
+ with person_logged_in(team.teamowner):
+ team.addMember(
+ team_member, team.teamowner, force_team_add=True)
+
+ self._test_edit_team_view_expected_subscription_vocab(
+ setup_team, OpenTeamSubscriptionPolicy)
+
def test_edit_team_view_save(self):
# A team can be edited and saved, including a name change, even if it
# is a private team and has a purged mailing list.
=== modified file 'lib/lp/registry/errors.py'
--- lib/lp/registry/errors.py 2011-11-08 10:47:31 +0000
+++ lib/lp/registry/errors.py 2011-11-08 10:47:33 +0000
@@ -28,7 +28,6 @@
import httplib
from lazr.restful.declarations import error_status
-from zope.schema.interfaces import ConstraintNotSatisfied
from zope.security.interfaces import Unauthorized
from lp.app.errors import NameLookupFailed
@@ -44,6 +43,11 @@
"""An attempt was made to link an open team to something."""
+@error_status(httplib.FORBIDDEN)
+class TeamSubscriptionPolicyError(ValueError):
+ """The team cannot have the specified TeamSubscriptionPolicy."""
+
+
@error_status(httplib.CONFLICT)
class NameAlreadyTaken(Exception):
"""The name given for a person is already in use by other person."""
@@ -144,30 +148,6 @@
@error_status(httplib.BAD_REQUEST)
-class TeamSubscriptionPolicyError(ConstraintNotSatisfied):
- """The team cannot have the specified TeamSubscriptionPolicy.
-
- The error can be raised because a super team or member team prevents
- this team from setting a specific policy. The error can also be
- raised if the team has an active PPA.
- """
-
- _default_message = "Team Subscription Policy Error"
-
- def __init__(self, message=None):
- if message is None:
- message = self._default_message
- self.message = message
-
- def doc(self):
- """See `Invalid`."""
- return self.message
-
- def __str__(self):
- return self.message
-
-
-@error_status(httplib.BAD_REQUEST)
class JoinNotAllowed(Exception):
"""User is not allowed to join a given team."""
=== modified file 'lib/lp/registry/interfaces/person.py'
--- lib/lp/registry/interfaces/person.py 2011-11-08 10:47:31 +0000
+++ lib/lp/registry/interfaces/person.py 2011-11-08 10:47:33 +0000
@@ -9,6 +9,7 @@
__all__ = [
'CLOSED_TEAM_POLICY',
+ 'ClosedTeamSubscriptionPolicy',
'IAdminPeopleMergeSchema',
'IAdminTeamMergeSchema',
'IHasStanding',
@@ -29,6 +30,7 @@
'ImmutableVisibilityError',
'NoSuchPerson',
'OPEN_TEAM_POLICY',
+ 'OpenTeamSubscriptionPolicy',
'PersonCreationRationale',
'PersonVisibility',
'PersonalStanding',
@@ -39,6 +41,7 @@
'validate_person',
'validate_person_or_closed_team',
'validate_public_person',
+ 'validate_subscription_policy',
]
from lazr.enum import (
@@ -70,12 +73,6 @@
Reference,
)
from lazr.restful.interface import copy_field
-from storm.expr import (
- And,
- Join,
- Select,
- Union,
- )
from zope.component import getUtility
from zope.formlib.form import NoInputData
from zope.interface import (
@@ -108,7 +105,6 @@
IHasMugshot,
IPrivacy,
)
-from canonical.launchpad.interfaces.lpstorm import IStore
from canonical.launchpad.interfaces.validation import validate_new_team_email
from canonical.launchpad.webapp.authorization import check_permission
from canonical.launchpad.webapp.interfaces import ILaunchpadApplication
@@ -163,7 +159,6 @@
StrippedTextLine,
)
from lp.services.worlddata.interfaces.language import ILanguage
-from lp.soyuz.enums import ArchiveStatus
from lp.translations.interfaces.hastranslationimports import (
IHasTranslationImports,
)
@@ -219,6 +214,33 @@
obj, attr, value, validate, error_class=OpenTeamLinkageError)
+def validate_subscription_policy(obj, attr, value):
+ """Validate the team subscription_policy."""
+ if value is None:
+ return None
+
+ # If we are just creating a new team, it can have any subscription policy.
+ if getattr(obj, '_SO_creating', True):
+ return value
+
+ team = obj
+ existing_subscription_policy = getattr(team, 'subscriptionpolicy', None)
+ if value == existing_subscription_policy:
+ return value
+ illegal_closed_transition = (
+ value in OpenTeamSubscriptionPolicy
+ and team.subscriptionPolicyMustBeClosed())
+ illegal_open_transition = (
+ value in ClosedTeamSubscriptionPolicy
+ and team.subscriptionPolicyMustBeOpen())
+ if illegal_closed_transition or illegal_open_transition:
+ raise TeamSubscriptionPolicyError(
+ "Cannot change subscription policy of %s (name=%s) from %s to %s"
+ % (team, getattr(team, 'name', None),
+ existing_subscription_policy, value))
+ return value
+
+
class PersonalStanding(DBEnumeratedType):
"""A person's standing.
@@ -463,6 +485,18 @@
""")
+class ClosedTeamSubscriptionPolicy(DBEnumeratedType):
+ MODERATED = TeamSubscriptionPolicy.MODERATED
+
+ RESTRICTED = TeamSubscriptionPolicy.RESTRICTED
+
+
+class OpenTeamSubscriptionPolicy(DBEnumeratedType):
+ OPEN = TeamSubscriptionPolicy.OPEN
+
+ DELEGATED = TeamSubscriptionPolicy.DELEGATED
+
+
OPEN_TEAM_POLICY = (
TeamSubscriptionPolicy.OPEN, TeamSubscriptionPolicy.DELEGATED)
@@ -541,107 +575,6 @@
super(PersonNameField, self)._validate(input)
-def team_subscription_policy_can_transition(team, policy):
- """Can the team can change its subscription policy?
-
- Returns True when the policy can change. or raises an error. OPEN teams
- cannot be members of MODERATED or RESTRICTED teams. OPEN teams
- cannot have PPAs. Changes from between OPEN and the two closed states
- can be blocked by team membership and team artifacts.
-
- :param team: The team to change.
- :param policy: The TeamSubsciptionPolicy to change to.
- :raises TeamSubsciptionPolicyError: Raised when a membership constrain
- or a team artifact prevents the policy from being set.
- """
- if team is None or policy == team.subscriptionpolicy:
- # The team is being initialized or the policy is not changing.
- return True
- elif policy in OPEN_TEAM_POLICY:
- # The team can be open if its super teams are open.
- for team in team.super_teams:
- if team.subscriptionpolicy in CLOSED_TEAM_POLICY:
- raise TeamSubscriptionPolicyError(
- "The team subscription policy cannot be %s because one "
- "or more if its super teams are not open." % policy)
- # The team can not be open if it has PPAs.
- for ppa in team.ppas:
- if ppa.status != ArchiveStatus.DELETED:
- raise TeamSubscriptionPolicyError(
- "The team subscription policy cannot be %s because it "
- "has one or more active PPAs." % policy)
- # Circular imports.
- from lp.bugs.model.bug import Bug
- from lp.bugs.model.bugsubscription import BugSubscription
- from lp.bugs.model.bugtask import BugTask
- # The team cannot be open if it is subscribed to or assigned to
- # private bugs.
- private_bugs_involved = IStore(Bug).execute(Union(
- Select(
- Bug.id,
- tables=(
- Bug,
- Join(BugSubscription, BugSubscription.bug_id == Bug.id)),
- where=And(
- Bug.private == True,
- BugSubscription.person_id == team.id)),
- Select(
- Bug.id,
- tables=(
- Bug,
- Join(BugTask, BugTask.bugID == Bug.id)),
- where=And(Bug.private == True, BugTask.assignee == team.id)),
- limit=1))
- if private_bugs_involved.rowcount:
- raise TeamSubscriptionPolicyError(
- "The team subscription policy cannot be %s because it is "
- "subscribed to or assigned to one or more private "
- "bugs." % policy)
- elif team.subscriptionpolicy in OPEN_TEAM_POLICY:
- # The team can become MODERATED or RESTRICTED if its member teams
- # are not OPEN.
- for member in team.activemembers:
- if member.subscriptionpolicy in OPEN_TEAM_POLICY:
- raise TeamSubscriptionPolicyError(
- "The team subscription policy cannot be %s because one "
- "or more if its member teams are Open." % policy)
- else:
- # The policy change is between MODERATED and RESTRICTED.
- pass
- return True
-
-
-class TeamSubsciptionPolicyChoice(Choice):
- """A valid team subscription policy."""
-
- def _getTeam(self):
- """Return the context if it is a team or None."""
- if IPerson.providedBy(self.context):
- return self.context
- else:
- return None
-
- def constraint(self, value):
- """See `IField`."""
- team = self._getTeam()
- policy = value
- try:
- return team_subscription_policy_can_transition(team, policy)
- except TeamSubscriptionPolicyError:
- return False
-
- def _validate(self, value):
- """Ensure the TeamSubsciptionPolicy is valid for state of the team.
-
- Returns True if the team can change its subscription policy to the
- `TeamSubscriptionPolicy`, otherwise raise TeamSubscriptionPolicyError.
- """
- team = self._getTeam()
- policy = value
- team_subscription_policy_can_transition(team, policy)
- super(TeamSubsciptionPolicyChoice, self)._validate(value)
-
-
class IPersonClaim(Interface):
"""The schema used by IPerson's +claim form."""
@@ -1947,7 +1880,7 @@
exported_as='team_description')
subscriptionpolicy = exported(
- TeamSubsciptionPolicyChoice(title=_('Subscription policy'),
+ Choice(title=_('Subscription policy'),
vocabulary=TeamSubscriptionPolicy,
default=TeamSubscriptionPolicy.MODERATED, required=True,
description=_(
@@ -1987,6 +1920,25 @@
"The date, according to team's default values, in "
"which a just-renewed membership will expire.")
+ def subscriptionPolicyMustBeClosed():
+ """ Return true if this team's subscription policy must be closed.
+
+ A closed subscription policy is MODERATED or RESTRICTED.
+ An closed subscription policy is required when:
+ - any of the team's super teams are closed.
+ - the team has any active PPAs
+ - it is subscribed or assigned to any private bugs
+ - it owns or is the security contact for any pillars
+ """
+
+ def subscriptionPolicyMustBeOpen():
+ """ Return true if this team's subscription policy must be open.
+
+ An open subscription policy is OPEN or DELEGATED.
+ An open subscription policy is required when:
+ - any of the team's sub (member) teams are open.
+ """
+
class ITeam(IPerson, ITeamPublic):
"""A group of people and other teams.
=== modified file 'lib/lp/registry/interfaces/role.py'
--- lib/lp/registry/interfaces/role.py 2011-07-21 22:42:14 +0000
+++ lib/lp/registry/interfaces/role.py 2011-11-08 10:47:33 +0000
@@ -122,6 +122,12 @@
Passed through to the same method in 'IPersonPublic'.
"""
+ def isPillarOwner():
+ """Is this person the owner of any pillar?"""
+
+ def isSecurityContact():
+ """Is this person the security contact of any pillar?"""
+
def isOwner(obj):
"""Is this person the owner of the object?"""
=== modified file 'lib/lp/registry/model/person.py'
--- lib/lp/registry/model/person.py 2011-10-27 03:02:10 +0000
+++ lib/lp/registry/model/person.py 2011-11-08 10:47:33 +0000
@@ -69,6 +69,7 @@
Or,
Select,
SQL,
+ Union,
Upper,
)
from storm.info import ClassAlias
@@ -229,23 +230,28 @@
)
from lp.registry.interfaces.person import (
CLOSED_TEAM_POLICY,
+ ClosedTeamSubscriptionPolicy,
ImmutableVisibilityError,
IPerson,
IPersonSet,
IPersonSettings,
ITeam,
+ OPEN_TEAM_POLICY,
+ OpenTeamSubscriptionPolicy,
PersonalStanding,
PersonCreationRationale,
PersonVisibility,
TeamMembershipRenewalPolicy,
TeamSubscriptionPolicy,
validate_public_person,
+ validate_subscription_policy,
)
from lp.registry.interfaces.personnotification import IPersonNotificationSet
from lp.registry.interfaces.persontransferjob import IPersonMergeJobSource
from lp.registry.interfaces.pillar import IPillarNameSet
from lp.registry.interfaces.product import IProduct
from lp.registry.interfaces.projectgroup import IProjectGroup
+from lp.registry.interfaces.role import IPersonRoles
from lp.registry.interfaces.ssh import (
ISSHKey,
ISSHKeySet,
@@ -585,8 +591,10 @@
default=TeamMembershipRenewalPolicy.NONE)
subscriptionpolicy = EnumCol(
dbName='subscriptionpolicy',
- enum=TeamSubscriptionPolicy,
- default=TeamSubscriptionPolicy.MODERATED)
+ enum=(ClosedTeamSubscriptionPolicy, OpenTeamSubscriptionPolicy,
+ TeamSubscriptionPolicy),
+ default=TeamSubscriptionPolicy.MODERATED,
+ storm_validator=validate_subscription_policy)
defaultrenewalperiod = IntCol(dbName='defaultrenewalperiod', default=None)
defaultmembershipperiod = IntCol(dbName='defaultmembershipperiod',
default=None)
@@ -1639,6 +1647,64 @@
EmailAddress.personID == self.id,
EmailAddress.status == status)
+ def subscriptionPolicyMustBeClosed(self):
+ """See `ITeam`"""
+ assert self.is_team, "This method must only be used for teams."
+
+ # Does this team own or is the security contact for any pillars.
+ roles = IPersonRoles(self)
+ if roles.isPillarOwner() or roles.isSecurityContact():
+ return True
+
+ # Does this team have any PPAs
+ for ppa in self.ppas:
+ if ppa.status != ArchiveStatus.DELETED:
+ return True
+
+ # Does this team have any super teams that are closed.
+ for team in self.super_teams:
+ if team.subscriptionpolicy in CLOSED_TEAM_POLICY:
+ return True
+
+ # Does this team subscribe or is assigned to any private bugs.
+ # Circular imports.
+ from lp.bugs.model.bug import Bug
+ from lp.bugs.model.bugsubscription import BugSubscription
+ from lp.bugs.model.bugtask import BugTask
+ # The team cannot be open if it is subscribed to or assigned to
+ # private bugs.
+ private_bugs_involved = IStore(Bug).execute(Union(
+ Select(
+ Bug.id,
+ tables=(
+ Bug,
+ Join(BugSubscription, BugSubscription.bug_id == Bug.id)),
+ where=And(
+ Bug.private == True,
+ BugSubscription.person_id == self.id)),
+ Select(
+ Bug.id,
+ tables=(
+ Bug,
+ Join(BugTask, BugTask.bugID == Bug.id)),
+ where=And(Bug.private == True, BugTask.assignee == self.id)),
+ limit=1))
+ if private_bugs_involved.rowcount:
+ return True
+
+ # We made it here, so let's return False.
+ return False
+
+ def subscriptionPolicyMustBeOpen(self):
+ """See `ITeam`"""
+ assert self.is_team, "This method must only be used for teams."
+
+ # The team must be open if any of it's members are open.
+ for member in self.activemembers:
+ if member.subscriptionpolicy in OPEN_TEAM_POLICY:
+ return True
+ return False
+
@property
def wiki_names(self):
"""See `IPerson`."""
=== modified file 'lib/lp/registry/model/personroles.py'
--- lib/lp/registry/model/personroles.py 2011-05-27 19:53:20 +0000
+++ lib/lp/registry/model/personroles.py 2011-11-08 10:47:33 +0000
@@ -6,12 +6,17 @@
__metaclass__ = type
__all__ = ['PersonRoles']
+from storm.expr import (
+ SQL,
+ With,
+ )
from zope.component import (
adapts,
getUtility,
)
from zope.interface import implements
+from canonical.launchpad.interfaces.lpstorm import IStore
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.registry.interfaces.person import IPerson
from lp.registry.interfaces.role import (
@@ -45,6 +50,58 @@
def id(self):
return self.person.id
+ def isPillarOwner(self):
+ """See IPersonRoles."""
+
+ with_sql = [
+ With("teams", SQL("""
+ SELECT team FROM TeamParticipation
+ WHERE TeamParticipation.person = %d
+ """ % self.person.id)),
+ With("owned_entities", SQL("""
+ SELECT Product.id
+ FROM Product
+ WHERE Product.owner IN (SELECT team FROM teams)
+ UNION ALL
+ SELECT Project.id
+ FROM Project
+ WHERE Project.owner IN (SELECT team FROM teams)
+ UNION ALL
+ SELECT Distribution.id
+ FROM Distribution
+ WHERE Distribution.owner IN (SELECT team FROM teams)
+ """))
+ ]
+ store = IStore(self.person)
+ rs = store.with_(with_sql).using("owned_entities").find(
+ SQL("count(*) > 0"),
+ )
+ return rs.one()
+
+ def isSecurityContact(self):
+ """See IPersonRoles."""
+ with_sql = [
+ With("teams", SQL("""
+ SELECT team FROM TeamParticipation
+ WHERE TeamParticipation.person = %d
+ """ % self.person.id)),
+ With("owned_entities", SQL("""
+ SELECT Product.id
+ FROM Product
+ WHERE Product.security_contact IN (SELECT team FROM teams)
+ UNION ALL
+ SELECT Distribution.id
+ FROM Distribution
+ WHERE Distribution.security_contact
+ IN (SELECT team FROM teams)
+ """))
+ ]
+ store = IStore(self.person)
+ rs = store.with_(with_sql).using("owned_entities").find(
+ SQL("count(*) > 0"),
+ )
+ return rs.one()
+
def isOwner(self, obj):
"""See IPersonRoles."""
return self.person.inTeam(obj.owner)
=== modified file 'lib/lp/registry/tests/test_errors.py'
--- lib/lp/registry/tests/test_errors.py 2011-11-08 10:47:31 +0000
+++ lib/lp/registry/tests/test_errors.py 2011-11-08 10:47:33 +0000
@@ -54,10 +54,10 @@
error_view = create_webservice_error_view(JoinNotAllowed())
self.assertEqual(BAD_REQUEST, error_view.status)
- def test_TeamSubscriptionPolicyError_bad_request(self):
+ def test_TeamSubscriptionPolicyError_forbidden(self):
error_view = create_webservice_error_view(
TeamSubscriptionPolicyError())
- self.assertEqual(BAD_REQUEST, error_view.status)
+ self.assertEqual(FORBIDDEN, error_view.status)
def test_TeamMembershipTransitionError_bad_request(self):
error_view = create_webservice_error_view(
=== modified file 'lib/lp/registry/tests/test_personroles.py'
--- lib/lp/registry/tests/test_personroles.py 2011-05-27 19:53:20 +0000
+++ lib/lp/registry/tests/test_personroles.py 2011-11-08 10:47:33 +0000
@@ -147,3 +147,43 @@
fake_attr = self.factory.getUniqueString()
roles = IPersonRoles(self.person)
self.assertRaises(AttributeError, roles.isOneOf, obj, [fake_attr])
+
+ def test_product_isOwner(self):
+ # Test isPillarOwner for products
+ person = self.factory.makePerson()
+ owner = self.factory.makePerson()
+ self.factory.makeProduct(owner=owner)
+ self.assertTrue(IPersonRoles(owner).isPillarOwner())
+ self.assertFalse(IPersonRoles(person).isPillarOwner())
+
+ def test_projectgroup_isOwner(self):
+ # Test isPillarOwner for project groups
+ person = self.factory.makePerson()
+ owner = self.factory.makePerson()
+ self.factory.makeProjectGroup(owner=owner)
+ self.assertTrue(IPersonRoles(owner).isPillarOwner())
+ self.assertFalse(IPersonRoles(person).isPillarOwner())
+
+ def test_distribution_isOwner(self):
+ # Test isPillarOwner for distributions
+ person = self.factory.makePerson()
+ owner = self.factory.makePerson()
+ self.factory.makeDistribution(owner=owner)
+ self.assertTrue(IPersonRoles(owner).isPillarOwner())
+ self.assertFalse(IPersonRoles(person).isPillarOwner())
+
+ def test_product_isSecurityContact(self):
+ # Test isSecurityContact for products
+ person = self.factory.makePerson()
+ contact = self.factory.makePerson()
+ self.factory.makeProduct(security_contact=contact)
+ self.assertTrue(IPersonRoles(contact).isSecurityContact())
+ self.assertFalse(IPersonRoles(person).isSecurityContact())
+
+ def test_distribution_isSecurityContact(self):
+ # Test isSecurityContact for distributions
+ person = self.factory.makePerson()
+ contact = self.factory.makePerson()
+ self.factory.makeDistribution(security_contact=contact)
+ self.assertTrue(IPersonRoles(contact).isSecurityContact())
+ self.assertFalse(IPersonRoles(person).isSecurityContact())
=== modified file 'lib/lp/registry/tests/test_team.py'
--- lib/lp/registry/tests/test_team.py 2011-10-24 05:15:58 +0000
+++ lib/lp/registry/tests/test_team.py 2011-11-08 10:47:33 +0000
@@ -13,19 +13,16 @@
from canonical.launchpad.database.emailaddress import EmailAddress
from canonical.launchpad.interfaces.emailaddress import IEmailAddressSet
from canonical.launchpad.interfaces.lpstorm import IMasterStore
-from canonical.testing.layers import (
- DatabaseFunctionalLayer,
- FunctionalLayer,
- )
+from canonical.testing.layers import DatabaseFunctionalLayer
from lp.registry.enum import PersonTransferJobType
from lp.registry.errors import (
JoinNotAllowed,
- TeamSubscriptionPolicyError,
- )
+ TeamSubscriptionPolicyError)
from lp.registry.interfaces.mailinglist import MailingListStatus
from lp.registry.interfaces.person import (
- IPersonSet,
+ CLOSED_TEAM_POLICY,
ITeamPublic,
+ OPEN_TEAM_POLICY,
PersonVisibility,
TeamMembershipRenewalPolicy,
TeamSubscriptionPolicy,
@@ -279,186 +276,106 @@
ITeamPublic['defaultmembershipperiod'].validate(3650)
-class TestTeamSubscriptionPolicyError(TestCaseWithFactory):
- """Test `TeamSubscriptionPolicyError` messages."""
-
- layer = FunctionalLayer
-
- def test_default_message(self):
- error = TeamSubscriptionPolicyError()
- self.assertEqual('Team Subscription Policy Error', error.message)
-
- def test_str(self):
- # The string is the error message.
- error = TeamSubscriptionPolicyError('a message')
- self.assertEqual('a message', str(error))
-
- def test_doc(self):
- # The doc() method returns the message. It is called when rendering
- # an error in the UI. eg structure error.
- error = TeamSubscriptionPolicyError('a message')
- self.assertEqual('a message', error.doc())
-
-
-class TeamSubscriptionPolicyBase(TestCaseWithFactory):
- """`TeamSubsciptionPolicyChoice` base test class."""
+class TestTeamSubscriptionPolicy(TestCaseWithFactory):
+ """Test whether teams must be open or closed."""
layer = DatabaseFunctionalLayer
- POLICY = None
- def setUpTeams(self, other_policy=None):
+ def setUpTeams(self, policy=None, other_policy=None):
+ if policy is None:
+ policy = TeamSubscriptionPolicy.MODERATED
if other_policy is None:
- other_policy = self.POLICY
- self.team = self.factory.makeTeam(subscription_policy=self.POLICY)
+ other_policy = TeamSubscriptionPolicy.MODERATED
+ self.team = self.factory.makeTeam(subscription_policy=policy)
self.other_team = self.factory.makeTeam(
subscription_policy=other_policy, owner=self.team.teamowner)
- self.field = ITeamPublic['subscriptionpolicy'].bind(self.team)
login_person(self.team.teamowner)
-
-class TestTeamSubscriptionPolicyChoiceCommon(TeamSubscriptionPolicyBase):
- """Test `TeamSubsciptionPolicyChoice` constraints."""
-
- # Any policy will work here, so we'll just pick one.
- POLICY = TeamSubscriptionPolicy.MODERATED
-
- def test___getTeam_with_team(self):
- # _getTeam returns the context team for team updates.
- self.setUpTeams()
- self.assertEqual(self.team, self.field._getTeam())
-
- def test___getTeam_with_person_set(self):
- # _getTeam returns the context person set for team creation.
- person_set = getUtility(IPersonSet)
- field = ITeamPublic['subscriptionpolicy'].bind(person_set)
- self.assertEqual(None, field._getTeam())
-
-
-class TestTeamSubscriptionPolicyChoiceModerated(TeamSubscriptionPolicyBase):
- """Test `TeamSubsciptionPolicyChoice` Moderated constraints."""
-
- POLICY = TeamSubscriptionPolicy.MODERATED
-
- def test_closed_team_with_closed_super_team_cannot_become_open(self):
+ def test_team_with_closed_super_team_cannot_be_open(self):
# The team cannot compromise the membership of the super team
- # by becoming open. The user must remove his team from the super team
+ # by being open. The user must remove his team from the super team
# first.
self.setUpTeams()
self.other_team.addMember(self.team, self.team.teamowner)
- self.assertFalse(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertRaises(
- TeamSubscriptionPolicyError, self.field.validate,
- TeamSubscriptionPolicy.OPEN)
+ self.assertTrue(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
- def test_closed_team_with_open_super_team_can_become_open(self):
- # The team can become open if its super teams are open.
+ def test_team_with_open_super_team_can_be_open(self):
+ # The team can be open if its super teams are open.
self.setUpTeams(other_policy=TeamSubscriptionPolicy.OPEN)
self.other_team.addMember(self.team, self.team.teamowner)
- self.assertTrue(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertEqual(
- None, self.field.validate(TeamSubscriptionPolicy.OPEN))
-
- def test_closed_team_can_change_to_another_closed_policy(self):
- # A closed team can change between the two closed polcies.
- self.setUpTeams()
- self.team.addMember(self.other_team, self.team.teamowner)
- super_team = self.factory.makeTeam(
- subscription_policy=TeamSubscriptionPolicy.MODERATED,
- owner=self.team.teamowner)
- super_team.addMember(self.team, self.team.teamowner)
- self.assertTrue(
- self.field.constraint(TeamSubscriptionPolicy.RESTRICTED))
- self.assertEqual(
- None, self.field.validate(TeamSubscriptionPolicy.RESTRICTED))
-
- def test_closed_team_with_active_ppas_cannot_become_open(self):
- # The team cannot become open if it has PPA because it compromises the
+ self.assertFalse(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
+
+ def test_team_with_active_ppas_cannot_be_open(self):
+ # The team cannot be open if it has PPA because it compromises
# the control of who can upload.
self.setUpTeams()
self.team.createPPA()
- self.assertFalse(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertRaises(
- TeamSubscriptionPolicyError, self.field.validate,
- TeamSubscriptionPolicy.OPEN)
+ self.assertTrue(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
- def test_closed_team_without_active_ppas_can_become_open(self):
- # The team can become if it has deleted PPAs.
+ def test_team_without_active_ppas_can_be_open(self):
+ # The team can be if it has deleted PPAs.
self.setUpTeams(other_policy=TeamSubscriptionPolicy.MODERATED)
ppa = self.team.createPPA()
ppa.delete(self.team.teamowner)
removeSecurityProxy(ppa).status = ArchiveStatus.DELETED
- self.assertTrue(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertEqual(
- None, self.field.validate(TeamSubscriptionPolicy.OPEN))
+ self.assertFalse(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
- def test_closed_team_with_private_bugs_cannot_become_open(self):
- # The team cannot become open if it is subscribed to private bugs.
+ def test_team_with_private_bugs_cannot_be_open(self):
+ # The team cannot be open if it is subscribed to private bugs.
self.setUpTeams()
bug = self.factory.makeBug(owner=self.team.teamowner, private=True)
with person_logged_in(self.team.teamowner):
bug.subscribe(self.team, self.team.teamowner)
- self.assertFalse(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertRaises(
- TeamSubscriptionPolicyError, self.field.validate,
- TeamSubscriptionPolicy.OPEN)
+ self.assertTrue(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
- def test_closed_team_with_private_bugs_assigned_cannot_become_open(self):
+ def test_team_with_private_bugs_assigned_cannot_be_open(self):
# The team cannot become open if it is assigned private bugs.
self.setUpTeams()
bug = self.factory.makeBug(owner=self.team.teamowner, private=True)
with person_logged_in(self.team.teamowner):
bug.default_bugtask.transitionToAssignee(self.team)
- self.assertFalse(
- self.field.constraint(TeamSubscriptionPolicy.OPEN))
- self.assertRaises(
- TeamSubscriptionPolicyError, self.field.validate,
- TeamSubscriptionPolicy.OPEN)
-
-
-class TestTeamSubscriptionPolicyChoiceRestrcted(
- TestTeamSubscriptionPolicyChoiceModerated):
- """Test `TeamSubsciptionPolicyChoice` Restricted constraints."""
-
- POLICY = TeamSubscriptionPolicy.RESTRICTED
-
-
-class TestTeamSubscriptionPolicyChoiceOpen(TeamSubscriptionPolicyBase):
- """Test `TeamSubsciptionPolicyChoice` Open constraints."""
-
- POLICY = TeamSubscriptionPolicy.OPEN
-
- def test_open_team_with_open_sub_team_cannot_become_closed(self):
- # The team cannot become closed if its membership will be
+ self.assertTrue(self.team.subscriptionPolicyMustBeClosed())
+ self.assertFalse(self.team.subscriptionPolicyMustBeOpen())
+
+ def test_team_with_open_sub_team_cannot_be_closed(self):
+ # The team cannot be closed if its membership will be
# compromised by an open subteam. The user must remove the subteam
- # first
+ # first.
+ self.setUpTeams(
+ policy=TeamSubscriptionPolicy.DELEGATED,
+ other_policy=TeamSubscriptionPolicy.OPEN)
+ self.team.addMember(self.other_team, self.team.teamowner)
+ self.assertTrue(self.team.subscriptionPolicyMustBeOpen())
+ self.assertFalse(self.team.subscriptionPolicyMustBeClosed())
+
+ def test_illegal_transition_to_open_subscription(self):
+ # Check that TeamSubscriptionPolicyError is raised when an attempt is
+ # made to set an illegal open subscription policy on a team.
self.setUpTeams()
- self.team.addMember(self.other_team, self.team.teamowner)
- self.assertFalse(
- self.field.constraint(TeamSubscriptionPolicy.MODERATED))
- self.assertRaises(
- TeamSubscriptionPolicyError, self.field.validate,
- TeamSubscriptionPolicy.MODERATED)
-
- def test_open_team_with_closed_sub_team_can_become_closed(self):
- # The team can become closed.
- self.setUpTeams(other_policy=TeamSubscriptionPolicy.MODERATED)
- self.team.addMember(self.other_team, self.team.teamowner)
- self.assertTrue(
- self.field.constraint(TeamSubscriptionPolicy.MODERATED))
- self.assertEqual(
- None, self.field.validate(TeamSubscriptionPolicy.MODERATED))
-
-
-class TestTeamSubscriptionPolicyChoiceDelegated(
- TestTeamSubscriptionPolicyChoiceOpen):
- """Test `TeamSubsciptionPolicyChoice` Delegated constraints."""
-
- POLICY = TeamSubscriptionPolicy.DELEGATED
+ self.team.createPPA()
+ for policy in OPEN_TEAM_POLICY:
+ self.assertRaises(
+ TeamSubscriptionPolicyError,
+ removeSecurityProxy(self.team).__setattr__,
+ "subscriptionpolicy", policy)
+
+ def test_illegal_transition_to_closed_subscription(self):
+ # Check that TeamSubscriptionPolicyError is raised when an attempt is
+ # made to set an illegal closed subscription policy on a team.
+ self.setUpTeams(
+ policy=TeamSubscriptionPolicy.DELEGATED,
+ other_policy=TeamSubscriptionPolicy.OPEN)
+ self.team.addMember(self.other_team, self.team.teamowner)
+ for policy in CLOSED_TEAM_POLICY:
+ self.assertRaises(
+ TeamSubscriptionPolicyError,
+ removeSecurityProxy(self.team).__setattr__,
+ "subscriptionpolicy", policy)
class TestVisibilityConsistencyWarning(TestCaseWithFactory):
Follow ups
