launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #28018
Re: [Merge] ~lgp171188/launchpad:enforce-stricter-permission-check-when-a-bug-is-locked-down into launchpad:master
Review: Approve
Thanks for these updates. There are a few more things to fix (especially issues around the garbo job), but I think it's nearly there now.
Diff comments:
> diff --git a/lib/lp/bugs/adapters/bugchange.py b/lib/lp/bugs/adapters/bugchange.py
> index ba0e99a..3a84c7c 100644
> --- a/lib/lp/bugs/adapters/bugchange.py
> +++ b/lib/lp/bugs/adapters/bugchange.py
> @@ -735,6 +741,80 @@ class CveUnlinkedFromBug(BugChangeBase):
> return {'text': "** CVE removed: %s" % self.cve.url}
>
>
> +class BugLocked(BugChangeBase):
> + """Used to represent the locking of a bug."""
> +
> + def __init__(self, when, person, old_status,
> + new_status, reason, **kwargs):
> + super().__init__(when, person)
> + self.old_status = old_status
> + self.new_status = new_status
> + self.reason = reason
> +
> + def getBugActivity(self):
> + """See `IBugChange'."""
> + activity = dict(
> + whatchanged='lock status',
> + oldvalue=str(self.old_status),
> + newvalue=str(self.new_status),
> + )
> + if self.reason:
> + activity['message'] = self.reason
> +
> + return activity
> +
> + def getBugNotification(self):
> + """See `IBugChange`."""
> + return {
> + 'text': ("** Bug metadata locked "
> + "and limited to project staff")
This indentation is rather confusing.
Perhaps this should also include the reason, if there is one?
> + }
> +
> +class BugUnlocked(BugChangeBase):
> + """Used to represent the unlocking of a bug."""
> +
> + def __init__(self, when, person, old_status, **kwargs):
> + super().__init__(when, person)
> + self.old_status = old_status
> +
> + def getBugActivity(self):
> + """See `IBugChange'."""
> + return dict(
> + whatchanged='lock status',
> + newvalue=str(BugLockStatus.UNLOCKED),
> + oldvalue=str(self.old_status),
> + )
> +
> + def getBugNotification(self):
> + """See `IBugChange`."""
> + return {
> + 'text': "** Bug unlocked"
> + }
> +
> +
> +class BugLockReasonSet(BugChangeBase):
> + """Used to represent setting of the Bug.lock_reason field."""
> +
> + def __init__(self, when, person, old_reason, new_reason, **kwargs):
> + super().__init__(when, person)
> + self.old_reason = old_reason if old_reason else 'unset'
> + self.new_reason = new_reason if new_reason else 'unset'
> +
> + def getBugActivity(self):
> + """See `IBugChange`."""
> + return dict(
> + whatchanged='lock reason',
> + oldvalue=self.old_reason,
> + newvalue=self.new_reason,
> + )
> +
> + def getBugNotification(self):
> + """See `IBugChange`."""
> + return {
> + 'text': "** Bug lock reason set - {}".format(self.new_reason)
A colon would be more idiomatic than a dash, I think.
Also, can we manage to have "Bug lock reason unset" here if the new reason is None? If you made the constructor just store the unmangled reasons and moved the `foo if foo else 'unset'` bit into `getBugActivity`, then that would be easier.
And finally, only if you agree, I wonder if "Bug lock reason changed" would be a little better?
> + }
> +
> +
> class BugTaskAttributeChange(AttributeChange):
> """Used to represent a change in a BugTask's attributes.
>
> diff --git a/lib/lp/bugs/browser/bugtask.py b/lib/lp/bugs/browser/bugtask.py
> index 1ae7e7c..d65f022 100644
> --- a/lib/lp/bugs/browser/bugtask.py
> +++ b/lib/lp/bugs/browser/bugtask.py
> @@ -2540,6 +2541,29 @@ class BugActivityItem:
> else:
> return_dict[key] = html_escape(return_dict[key])
>
> + elif attribute == 'lock status':
> + message = " "
> + if self.newvalue == str(BugLockStatus.COMMENT_ONLY):
Maybe move `message = " "` inside this conditional, since it's only used inside this conditional?
> + detail = (
> + 'Metadata changes locked{}and '
> + 'limited to project staff'
> + )
> + if self.message:
> + message = " ({}) ".format(html_escape(self.message))
> + return detail.format(message)
> + else:
> + return 'Metadata changes unlocked'
> + elif attribute == 'lock reason':
> + if self.newvalue != "unset" and self.oldvalue != "unset":
> + # return a proper message with old and new values
> + return "{} → {}".format(
`→` means the same thing and is clearer.
> + html_escape(self.oldvalue), html_escape(self.newvalue)
> + )
> + pass
> + elif self.newvalue != "unset":
> + return "{}".format(self.newvalue)
> + else:
> + return "Unset"
> elif attribute == 'milestone':
> for key in return_dict:
> if return_dict[key] is None:
> diff --git a/lib/lp/bugs/configure.zcml b/lib/lp/bugs/configure.zcml
> index 5b46af7..076294e 100644
> --- a/lib/lp/bugs/configure.zcml
> +++ b/lib/lp/bugs/configure.zcml
> @@ -740,6 +740,12 @@
> tags
> title
> description"/>
> + <require
> + permission="launchpad.Moderate"
> + set_attributes="lock_reason"/>
> + <require
> + permission="launchpad.Moderate"
> + interface="lp.bugs.interfaces.bug.IBugModerate"/>
You can combine these, and we normally do:
<require
permission="launchpad.Moderate"
interface="lp.bugs.interfaces.bug.IBugModerate"
set_attributes="lock_reason"/>
> </class>
> <adapter
> for="lp.bugs.interfaces.bug.IBug"
> diff --git a/lib/lp/bugs/model/bug.py b/lib/lp/bugs/model/bug.py
> index f7c588c..eb413c7 100644
> --- a/lib/lp/bugs/model/bug.py
> +++ b/lib/lp/bugs/model/bug.py
> @@ -387,6 +408,9 @@ class Bug(SQLBase, InformationTypeMixin):
> heat = IntCol(notNull=True, default=0)
> heat_last_updated = UtcDateTimeCol(default=None)
> latest_patch_uploaded = UtcDateTimeCol(default=None)
> + lock_status = DBEnum(
> + enum=BugLockStatus, allow_none=True, default=BugLockStatus.UNLOCKED)
I'd recommend temporarily calling this `_lock_status`, and adding:
@property
def lock_status(self):
return BugLockStatus.UNLOCKED if self._lock_status is None else self._lock_status
@lock_status.setter
def lock_status(self, value):
self._lock_status = value
This does mean that the garbo job will have to reference `Bug._lock_status` in its Storm query rather than `Bug.lock_status`, but otherwise everything should work. The reason for this is that it allows you to write code that behaves as if the garbo job has finished running without having to check for None at every call site. (The `lock`, `unlock`, and `setLockReason` methods will all currently misbehave in various ways if the lock status is None, and it's best to fix this up centrally using a property.)
> + lock_reason = StringCol(notNull=False, default=None)
>
> @property
> def linked_branches(self):
> @@ -2198,6 +2222,63 @@ class Bug(SQLBase, InformationTypeMixin):
> BugActivity.datechanged <= end_date)
> return activity_in_range
>
> + def lock(self, who, status, reason=None):
> + """See `IBug`."""
> + if status == self.lock_status:
> + raise CannotLockBug(
> + "This bug is already locked "
> + "with the lock status '{}'.".format(
> + str(status)
> + )
You don't need the `str()`, as `str.format()` does that automatically (https://docs.python.org/3/library/string.html#formatspec).
> + )
> + else:
> + old_lock_status = self.lock_status
> + self.lock_status = BugLockStatus.items[status.value]
> + if reason:
> + self.lock_reason = reason
> +
> + self.addChange(
> + BugLocked(
> + when=UTC_NOW,
> + person=who,
> + old_status=old_lock_status,
> + new_status=self.lock_status,
> + reason=reason
> + )
> + )
> +
> + def unlock(self, who):
> + """See `IBug`."""
> + if self.lock_status != BugLockStatus.UNLOCKED:
> + old_lock_status = self.lock_status
> + self.lock_status = BugLockStatus.UNLOCKED
> + self.lock_reason = None
> +
> + self.addChange(
> + BugUnlocked(
> + when=UTC_NOW,
> + person=who,
> + old_status=old_lock_status
> + )
> + )
> +
> + def setLockReason(self, reason, who):
> + """See `IBug`."""
> + if self.lock_status == BugLockStatus.UNLOCKED:
> + raise CannotSetLockReason(
> + "Lock reason cannot be set for an unlocked bug."
> + )
> + if self.lock_reason != reason:
> + old_reason = self.lock_reason
> + self.lock_reason = reason
> + self.addChange(
> + BugLockReasonSet(
> + when=UTC_NOW,
> + person=who,
> + old_reason=old_reason,
> + new_reason=reason
> + )
> + )
>
> @ProxyFactory
> def get_also_notified_subscribers(
> diff --git a/lib/lp/bugs/security.py b/lib/lp/bugs/security.py
> index b60b426..af93acb 100644
> --- a/lib/lp/bugs/security.py
> +++ b/lib/lp/bugs/security.py
> @@ -173,18 +187,37 @@ class EditBug(AuthorizationBase):
> # If the user seems generally legitimate, let them through.
> self.forwardCheckAuthenticated(
> user, permission='launchpad.AnyLegitimatePerson') or
> - # The bug reporter can always edit their own bug.
I'd be inclined to change this comment rather than removing it, just to avoid making it look as though it goes with the comment a few lines up. Perhaps this?
# The bug reporter can edit their own bug if it is unlocked.
> user.inTeam(self.obj.owner) or
> - # Users with relevant roles can edit the bug.
> - user.in_admin or user.in_commercial_admin or
> - user.in_registry_experts or
> - _has_any_bug_role(user, self.obj.bugtasks))
> + in_allowed_roles())
>
> def checkUnauthenticated(self):
> """Never allow unauthenticated users to edit a bug."""
> return False
>
>
> +class ModerateBug(AuthorizationBase):
> + """Security adapter for moderating bugs.
> +
> + This is used for operations like locking and unlocking a bug to the
> + relevant roles.
> + """
> + permission = 'launchpad.Moderate'
> + usedfor = IBug
> +
> + def checkAuthenticated(self, user):
> + if not self.forwardCheckAuthenticated(
> + user, permission='launchpad.Append'):
> + # The user cannot even see the bug.
> + return False
> +
> + return (
> + user.in_admin or
> + user.in_commercial_admin or
> + user.in_registry_experts or
> + _has_any_bug_role(user, self.obj.bugtasks)
> + )
> +
> +
> class PublicToAllOrPrivateToExplicitSubscribersForBug(AuthorizationBase):
> permission = 'launchpad.View'
> usedfor = IBug
> diff --git a/lib/lp/bugs/tests/test_bug.py b/lib/lp/bugs/tests/test_bug.py
> index 0c1e24b..4af27cc 100644
> --- a/lib/lp/bugs/tests/test_bug.py
> +++ b/lib/lp/bugs/tests/test_bug.py
> @@ -410,3 +423,426 @@ class TestBugPermissions(TestCaseWithFactory, KarmaTestMixin):
> person)
> with person_logged_in(person):
> self.assertTrue(checkPermission('launchpad.Edit', self.bug))
> +
> +
> +class TestBugLocking(TestCaseWithFactory):
> + """
> + Tests for the bug locking functionality.
> + """
> +
> + layer = DatabaseFunctionalLayer
> +
> + def setUp(self):
> + super().setUp()
> + self.person = self.factory.makePerson()
> + self.target = self.factory.makeProduct()
> +
> + def test_bug_lock_status_lock_reason_default_values(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + self.assertEqual(BugLockStatus.UNLOCKED, bug.lock_status)
> + self.assertIsNone(bug.lock_reason)
> +
> + def test_bug_locking_when_bug_already_locked(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(who=self.target.owner, status=BugLockStatus.COMMENT_ONLY)
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> + self.assertRaises(
> + CannotLockBug,
> + bug.lock,
> + who=self.target.owner,
> + status=BugLockStatus.COMMENT_ONLY
> + )
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> +
> + def test_bug_locking_with_a_reason_works(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.person,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot',
> + )
> + self.assertEqual('too hot', bug.lock_reason)
> +
> + def test_updating_bug_lock_reason_not_set_before(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(who=self.person, status=BugLockStatus.COMMENT_ONLY)
> + self.assertIsNone(bug.lock_reason)
> + bug.setLockReason('too hot', who=self.target.owner)
> + self.assertEqual('too hot', bug.lock_reason)
> +
> + def test_updating_existing_bug_lock_reason_to_none(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.person,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot',
> + )
> + self.assertEqual('too hot', bug.lock_reason)
> + bug.setLockReason(None, who=self.target.owner)
> + self.assertIsNone(bug.lock_reason)
> +
> + def test_bug_unlocking_clears_the_reason(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.person,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot',
> + )
> + bug.unlock(who=self.person)
> + self.assertIsNone(bug.lock_reason)
> +
> + def test_bug_locking_unlocking_adds_bug_activity_entries(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + self.assertEqual(1, bug.activity.count())
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.target.owner,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot'
> + )
> + self.assertEqual(2, bug.activity.count())
> + self.assertThat(
> + bug.activity[1],
> + MatchesStructure.byEquality(
> + person=self.target.owner,
> + whatchanged='lock status',
> + oldvalue=str(BugLockStatus.UNLOCKED),
> + newvalue=str(BugLockStatus.COMMENT_ONLY),
> + )
> + )
> + bug.unlock(who=self.target.owner)
> + self.assertEqual(3, bug.activity.count())
> + self.assertThat(
> + bug.activity[2],
> + MatchesStructure.byEquality(
> + person=self.target.owner,
> + whatchanged='lock status',
> + oldvalue=str(BugLockStatus.COMMENT_ONLY),
> + newvalue=str(BugLockStatus.UNLOCKED),
> + )
> + )
> +
> + def test_cannot_set_lock_reason_for_an_unlocked_bug(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + self.assertRaises(
> + CannotSetLockReason,
> + bug.setLockReason,
> + 'too hot',
> + who=self.target.owner
> + )
> +
> + def test_edit_permission_restrictions_when_a_bug_is_locked(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + another_person = self.factory.makePerson()
> +
> + with person_logged_in(self.target.owner):
> + bug.lock(who=self.target.owner, status=BugLockStatus.COMMENT_ONLY)
> +
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> +
> + # A user without the relevant role cannot edit a locked bug.
> + with person_logged_in(another_person):
> + self.assertFalse(checkPermission('launchpad.Edit', bug))
> +
> + # The bug reporter cannot edit a locked bug.
> + with person_logged_in(self.person):
> + self.assertFalse(checkPermission('launchpad.Edit', bug))
> +
> + # Target driver can can edit a locked bug.
Duplicate "can".
> + new_person = self.factory.makePerson()
> + removeSecurityProxy(bug.default_bugtask.target).driver = new_person
> + with person_logged_in(new_person):
> + self.assertTrue(checkPermission('launchpad.Edit', bug))
> +
> + # Admins can edit a locked bug.
> + with admin_logged_in():
> + self.assertTrue(checkPermission('launchpad.Edit', bug))
> +
> + # Commercial admins can edit a locked bug.
> + with celebrity_logged_in('commercial_admin'):
> + self.assertTrue(checkPermission('launchpad.Edit', bug))
> +
> + # Registry experts can edit a locked bug.
> + with celebrity_logged_in('registry_experts'):
> + self.assertTrue(checkPermission('launchpad.Edit', bug))
> +
> +
> + def test_only_those_with_moderate_permission_can_lock_unlock_a_bug(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + another_person = self.factory.makePerson()
> +
> + # Unauthenticated person cannot moderate a bug.
> + self.assertFalse(checkPermission('launchpad.Moderate', bug))
> +
> + # A user without the relevant role cannot moderate a bug.
> + with person_logged_in(another_person):
> + self.assertFalse(checkPermission('launchpad.Moderate', bug))
> +
> + # The bug reporter cannot moderate a bug.
> + with person_logged_in(self.person):
> + self.assertFalse(checkPermission('launchpad.Moderate', bug))
> +
> + # Admins can moderate a bug.
> + with admin_logged_in():
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> + # Commercial admins can moderate a bug.
> + with celebrity_logged_in('commercial_admin'):
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> + # Registry experts can moderate a bug.
> + with celebrity_logged_in('registry_experts'):
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> + # Target owner can moderate a bug.
> + with person_logged_in(
> + removeSecurityProxy(bug.default_bugtask.target).owner
> + ):
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> + # Target driver can moderate a bug.
> + new_person = self.factory.makePerson()
> + removeSecurityProxy(bug.default_bugtask.target).driver = new_person
> + with person_logged_in(new_person):
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> + yet_another_person = self.factory.makePerson()
> + removeSecurityProxy(
> + bug.default_bugtask.target
> + ).bug_supervisor = yet_another_person
> + with person_logged_in(yet_another_person):
> + self.assertTrue(checkPermission('launchpad.Moderate', bug))
> +
> +
> +class TestBugLockingWebService(TestCaseWithFactory):
> + """Tests for the bug locking and unlocking web service methods."""
> +
> + layer = DatabaseFunctionalLayer
> +
> + def setUp(self):
> + super().setUp()
> + self.person = self.factory.makePerson()
> + self.target = self.factory.makeProduct()
> +
> + def test_bug_lock_status_invalid_values(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + invalid_values_list = ["test", 1.23, 123, "Unlocked"]
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + bug_url = api_url(bug)
> + webservice.default_api_version = "devel"
> + for invalid_value in invalid_values_list:
> + response = webservice.named_post(
> + bug_url, 'lock', status=invalid_value
> + )
> + self.assertEqual(400, response.status)
> +
> + def test_who_value_for_lock_is_correctly_set(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + self.assertEqual(bug.activity.count(), 1)
> + self.assertEqual(BugLockStatus.UNLOCKED, bug.lock_status)
> +
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = "devel"
> +
> + response = webservice.named_post(
> + bug_url, 'lock', status=str(BugLockStatus.COMMENT_ONLY),
For webservice tests I normally write out the string (i.e. "Comment-only" here). Although this feels like duplication, the reason for it is that it reminds us that the enumeration item titles are part of the API: changing the titles might break people's scripts, and so it makes sense for us to be reminded of that by tests that fail if we change the titles.
> + )
> + self.assertEqual(200, response.status)
> +
> + with person_logged_in(ANONYMOUS):
> + self.assertEqual(2, bug.activity.count())
> + self.assertThat(
> + bug.activity[1],
> + MatchesStructure.byEquality(
> + person=self.target.owner,
> + whatchanged='lock status',
> + oldvalue=str(BugLockStatus.UNLOCKED),
> + newvalue=str(BugLockStatus.COMMENT_ONLY)
> + )
> + )
> +
> + def test_who_value_for_unlock_is_correctly_set(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + self.assertEqual(1, bug.activity.count())
> + with person_logged_in(self.target.owner):
> + bug.lock(who=self.target.owner, status=BugLockStatus.COMMENT_ONLY)
> + self.assertEqual(2, bug.activity.count())
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = "devel"
> +
> + response = webservice.named_post(
> + bug_url, 'unlock'
> + )
> + self.assertEqual(200, response.status)
> + with person_logged_in(ANONYMOUS):
> + self.assertEqual(3, bug.activity.count())
> + self.assertThat(
> + bug.activity[2],
> + MatchesStructure.byEquality(
> + person=self.target.owner,
> + whatchanged='lock status',
> + oldvalue=str(BugLockStatus.COMMENT_ONLY),
> + newvalue=str(BugLockStatus.UNLOCKED),
> + )
> + )
> +
> + def test_lock_status_lock_reason_values_unlocked_bug(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + bug_url = api_url(bug)
> +
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.get(bug_url)
> + self.assertEqual(200, response.status)
> + response_json = response.jsonBody()
> + self.assertEqual(
> + str(BugLockStatus.UNLOCKED),
> + response_json['lock_status']
> + )
> + self.assertIsNone(response_json['lock_reason'])
> +
> + def test_lock_status_lock_reason_values_after_locking(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + bug_url = api_url(bug)
> +
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.target.owner,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot'
> + )
> +
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.get(bug_url)
> + self.assertEqual(200, response.status)
> + response_json = response.jsonBody()
> + self.assertEqual(
> + str(BugLockStatus.COMMENT_ONLY),
> + response_json['lock_status']
> + )
> + self.assertEqual(
> + response_json['lock_reason'],
> + 'too hot'
> + )
> +
> + def test_setting_lock_reason_for_an_unlocked_bug(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.patch(
> + bug_url, "application/json",
> + json.dumps({'lock_reason': 'too hot'})
> + )
> + self.assertThat(
> + response,
> + MatchesStructure.byEquality(
> + status=400,
> + body=b'Lock reason cannot be set for an unlocked bug.'
> + )
> + )
> +
> + def test_setting_lock_reason_for_a_locked_bug_without_a_reason(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(who=self.target.owner, status=BugLockStatus.COMMENT_ONLY)
> +
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> + self.assertIsNone(bug.lock_reason)
> +
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.patch(
> + bug_url, "application/json",
> + json.dumps({'lock_reason': 'too hot'})
> + )
> + self.assertEqual(209, response.status)
> + self.assertEqual('too hot', response.jsonBody()['lock_reason'])
> +
> +
> + def test_setting_lock_reason_for_a_locked_bug_with_a_reason(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.target.owner,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot'
> + )
> +
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> + self.assertEqual('too hot', bug.lock_reason)
> +
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.patch(
> + bug_url, "application/json",
> + json.dumps({'lock_reason': 'too hot!'})
> + )
> + self.assertEqual(209, response.status)
> + self.assertEqual('too hot!', response.jsonBody()['lock_reason'])
> +
> + def test_removing_lock_reason_for_a_locked_bug_with_a_reason(self):
> + bug = self.factory.makeBug(owner=self.person, target=self.target)
> + with person_logged_in(self.target.owner):
> + bug.lock(
> + who=self.target.owner,
> + status=BugLockStatus.COMMENT_ONLY,
> + reason='too hot'
> + )
> +
> + self.assertEqual(BugLockStatus.COMMENT_ONLY, bug.lock_status)
> + self.assertEqual('too hot', bug.lock_reason)
> +
> + bug_url = api_url(bug)
> + webservice = webservice_for_person(
> + self.target.owner,
> + permission=OAuthPermission.WRITE_PRIVATE
> + )
> + webservice.default_api_version = 'devel'
> +
> + response = webservice.patch(
> + bug_url, "application/json",
> + json.dumps({'lock_reason': None})
> + )
> + self.assertEqual(209, response.status)
> + self.assertEqual(None, response.jsonBody()['lock_reason'])
> diff --git a/lib/lp/scripts/garbo.py b/lib/lp/scripts/garbo.py
> index 5eabeb6..8b865fd 100644
> --- a/lib/lp/scripts/garbo.py
> +++ b/lib/lp/scripts/garbo.py
> @@ -1762,6 +1763,37 @@ class PopulateSnapBuildStoreRevision(TunableLoop):
> transaction.commit()
>
>
> +class PopulateBugLockStatusDefaultUnlocked(TunableLoop):
> + """
> + Populates Bug.lock_status to BugLockStatus.UNLOCKED if not set
> + """
> +
> + maximum_chunk_size = 5000
> +
> + def __init__(self, log, abort_time=None):
> + super().__init__(log, abort_time)
> + self.store = IMasterStore(Bug)
> + self.start_at = 1
> +
> + def findBugsLockStatusNone(self):
> + return self.store.find(
> + Bug,
> + Bug.id >= self.start_at,
> + Bug.lock_status == None,
> + ).order_by(Bug.id)
> +
> + def isDone(self):
> + return self.findBugsLockStatusNone().is_empty()
> +
> + def __call__(self, chunk_size):
> + bugs = self.findBugsLockStatusNone()[:chunk_size]
> + bugs.set(lock_status=BugLockStatus.UNLOCKED)
> + bugs = list(bugs)
> + if bugs:
> + self.start_at = bugs[-1].id + 1
`TunableLoop.run` guarantees that the result set of bugs is non-empty here (by checking `isDone` first), so I think you can drop both `bugs = list(bugs)` and the `if bugs:` check.
However, does this work as intended? When you evaluate `list(bugs)` (in your version) or `bugs[-1]` (with my adjustments), that's going to cause Storm to execute a `SELECT` query, which happens after the `UPDATE` executed by `bugs.set(...)`, and so it'll set `start_at` to the wrong ID. I think you may also run into https://bugs.launchpad.net/storm/+bug/820290, causing `bugs.set` to do the wrong thing. Admittedly I haven't verified this (you could do so using `LP_DEBUG_SQL=1` to look at the actual SQL queries being executed), but I'd be inclined to write this more defensively, along the lines of the following:
bug_ids = [bug.id for bug in self.findBugsLockStatusNone()[:chunk_size]]
self.store.find(Bug, Bug.id.is_in(bug_ids)).set(lock_status=BugLockStatus.UNLOCKED)
self.start_at = bug_ids[-1] + 1
You could then also make this more efficient (pulling less data back from the database) by changing `findBugsLockStatusNone` to `findBugIDsLockStatusNone` and making it query for `Bug.id` rather than just `Bug`. The `Bug` table is wide enough that this is probably worthwhile.
> + transaction.commit()
> +
> +
> class BaseDatabaseGarbageCollector(LaunchpadCronScript):
> """Abstract base class to run a collection of TunableLoops."""
> script_name = None # Script name for locking and database user. Override.
> @@ -2054,6 +2086,7 @@ class DailyDatabaseGarbageCollector(BaseDatabaseGarbageCollector):
> ObsoleteBugAttachmentPruner,
> OldTimeLimitedTokenDeleter,
> PopulateSnapBuildStoreRevision,
> + PopulateBugLockStatusDefaultUnlocked,
Could you keep this list sorted?
> POTranslationPruner,
> PreviewDiffPruner,
> ProductVCSPopulator,
--
https://code.launchpad.net/~lgp171188/launchpad/+git/launchpad/+merge/414097
Your team Launchpad code reviewers is requested to review the proposed merge of ~lgp171188/launchpad:enforce-stricter-permission-check-when-a-bug-is-locked-down into launchpad:master.
References
