jenkins-bot has submitted this change and it was merged. ( https://gerrit.wikimedia.org/r/565624 )
Change subject: [IMPR] Replace @must_be with @need_right ......................................................................
[IMPR] Replace @must_be with @need_right
After dropping of user/sysop dualism, introduce a need_right decorator in favour of must_be decorator which primarily checks group membership. This decorator checks user right to be needed before a method will be executed.
- deprecate "right" parameter in must_be - do not try to login with must_be decorator. login() method does not have the sysop parameter any longer - must_be enables checking for other user groups than just sysop and user - must_be and need_right checks for "steward" user group if any request is taken for obsolete wikis - replace all must_be usage with the new need_right decorator - tests added and modified
Bug: T57296 Bug: T71283 Bug: T233608 Change-Id: I61622c2a7cfbcd8a6d33b5fdbaeefb355c47e3e9 --- M pywikibot/site.py M tests/dry_site_tests.py 2 files changed, 195 insertions(+), 75 deletions(-)
Approvals: Huji: Looks good to me, approved jenkins-bot: Verified
diff --git a/pywikibot/site.py b/pywikibot/site.py index f2bd0b3..e16d78f 100644 --- a/pywikibot/site.py +++ b/pywikibot/site.py @@ -1285,33 +1285,31 @@ return self.getUrl(address, data=data)
-def must_be(group=None, right=None): +@deprecated_args(right=None) +def must_be(group=None): """Decorator to require a certain user status when method is called.
- @param group: The group the logged in user should belong to - this parameter can be overridden by + @param group: The group the logged in user should belong to. + This parameter can be overridden by keyword argument 'as_group'. - @type group: str ('user' or 'sysop') - @param right: The rights the logged in user should have. - + @type group: str @return: method decorator + @raises UserRightsError: user is not part of the required user group. """ def decorator(fn): def callee(self, *args, **kwargs): grp = kwargs.pop('as_group', group) - if self.obsolete and not self.has_group('steward'): - raise UserRightsError('Site {} has been closed. Only steward ' - 'can perform requested action.' - .format(self.sitename)) - if right is not None: - if self.has_right(right): - return fn(self, *args, **kwargs) - if grp == 'user': - self.login(False) - elif grp == 'sysop': - self.login(True) - else: - raise Exception('Not implemented') + if self.obsolete: + if not self.has_group('steward'): + raise UserRightsError( + 'Site {} has been closed. Only steward ' + 'can perform requested action.' + .format(self.sitename)) + + elif not self.has_group(grp): + raise UserRightsError('User "{}" is not part of the required ' + 'user group "{}"' + .format(self.user(), grp))
return fn(self, *args, **kwargs)
@@ -1319,7 +1317,38 @@ return fn
manage_wrapping(callee, fn) + return callee
+ return decorator + + +def need_right(right=None): + """Decorator to require a certain user right when method is called. + + @param right: The right the logged in user should have. + @type right: str + @return: method decorator + @raises UserRightsError: user has insufficient rights. + """ + def decorator(fn): + def callee(self, *args, **kwargs): + if self.obsolete: + if not self.has_group('steward'): + raise UserRightsError( + 'Site {} has been closed. Only steward ' + 'can perform requested action.' + .format(self.sitename)) + + elif right is not None and not self.has_right(right): + raise UserRightsError('User "{}" does not have required ' + 'user right "{}"' + .format(self.user, right)) + return fn(self, *args, **kwargs) + + if not __debug__: + return fn + + manage_wrapping(callee, fn) return callee
return decorator @@ -2271,13 +2300,11 @@ def has_right(self, right): """Return true if and only if the user has a specific right.
- Possible values of 'right' may vary depending on wiki settings, - but will usually include: - - * Actions: edit, move, delete, protect, upload - * User levels: autoconfirmed, sysop, bot - + Possible values of 'right' may vary depending on wiki settings. U{https://www.mediawiki.org/wiki/API:Userinfo%7D + + @param right: a specific right to be validated + @type right: str """ return right.lower() in self._userinfo['rights']
@@ -5210,7 +5237,7 @@ } _ep_text_overrides = {'appendtext', 'prependtext', 'undo'}
- @must_be(group='user') + @need_right('edit') def editpage(self, page, summary=None, minor=True, notminor=False, bot=True, recreate=True, createonly=False, nocreate=False, watch=None, **kwargs): @@ -5430,7 +5457,7 @@ 'destination revisions of {dest}' }
- @must_be(group='sysop', right='mergehistory') + @need_right('mergehistory') def merge_history(self, source, dest, timestamp=None, reason=None): """Merge revisions from one page into another.
@@ -5549,7 +5576,7 @@ '[[%(oldtitle)s]]', }
- @must_be(group='user') + @need_right('move') def movepage(self, page, newtitle, summary, movetalk=True, noredirect=False): """Move a Page to a new title. @@ -5652,7 +5679,7 @@ 'Page [[%(title)s]] already rolled back; action aborted.', } # other errors shouldn't arise because we check for those errors
- @must_be('user') + @need_right('rollback') def rollbackpage(self, page, **kwargs): """Roll back page to version before last user's edits.
@@ -5717,7 +5744,7 @@ 'Revision may not exist or was already undeleted.' } # other errors shouldn't occur because of pre-submission checks
- @must_be(group='sysop', right='delete') + @need_right('delete') @deprecate_arg('summary', 'reason') def deletepage(self, page, reason): """Delete page from the wiki. Requires appropriate privilege level. @@ -5755,7 +5782,7 @@ finally: self.unlock_page(page)
- @must_be(group='sysop', right='undelete') + @need_right('undelete') @deprecate_arg('summary', 'reason') def undelete_page(self, page, reason, revisions=None): """Undelete page from the wiki. Requires appropriate privilege level. @@ -5828,7 +5855,7 @@ # implemented in b73b5883d486db0e9278ef16733551f28d9e096d return set(self.siteinfo.get('restrictions')['levels'])
- @must_be(group='sysop', right='protect') + @need_right('protect') @deprecate_arg('summary', 'reason') def protect(self, page, protections, reason, expiry=None, **kwargs): """(Un)protect a wiki page. Requires administrator status. @@ -5897,7 +5924,7 @@ "The revision %(revid)s can't be patrolled as it's too old." }
- @must_be(group='user') + @need_right('patrol') @deprecated_args(token=None) def patrol(self, rcid=None, revid=None, revision=None): """Return a generator of patrolled pages. @@ -5986,7 +6013,7 @@
yield result['patrol']
- @must_be(group='sysop', right='block') + @need_right('block') def blockuser(self, user, expiry, reason, anononly=True, nocreate=True, autoblock=True, noemail=False, reblock=False, allowusertalk=False): @@ -6044,7 +6071,7 @@ data = req.submit() return data
- @must_be(group='sysop', right='block') + @need_right('unblock') def unblockuser(self, user, reason=None): """ Remove the block for the user. @@ -6064,7 +6091,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('editmywatchlist') def watch(self, pages, unwatch=False): """Add or remove pages from watchlist.
@@ -6104,7 +6131,7 @@ return False return True
- @must_be(group='user') + @need_right('editmywatchlist') @deprecated('Site().watch', since='20160102') def watchpage(self, page, unwatch=False): """ @@ -6131,7 +6158,7 @@ return False return ('unwatched' if unwatch else 'watched') in result['watch']
- @must_be(group='user') + @need_right('purge') def purgepages( self, pages, forcelinkupdate=False, forcerecursivelinkupdate=False, converttitles=False, redirects=False @@ -6211,7 +6238,7 @@ """ return self._get_titles_with_hash(hash_found)
- @must_be(group='user') + @need_right('edit') def is_uploaddisabled(self): """Return True if upload is disabled on site.
@@ -7269,7 +7296,7 @@ data = req.submit() return data['flow']['view-post']['result']['topic']
- @must_be('user') + @need_right('edit') @need_extension('Flow') def create_new_topic(self, page, title, content, format): """ @@ -7294,7 +7321,7 @@ data = req.submit() return data['flow']['new-topic']['committed']['topiclist']
- @must_be('user') + @need_right('edit') @need_extension('Flow') def reply_to_post(self, page, reply_to_uuid, content, format): """Reply to a post on a Flow topic. @@ -7318,7 +7345,7 @@ data = req.submit() return data['flow']['reply']['committed']['topic']
- @must_be('user', 'flow-lock') + @need_right('flow-lock') @need_extension('Flow') def lock_topic(self, page, lock, reason): """ @@ -7342,7 +7369,7 @@ data = req.submit() return data['flow']['lock-topic']['committed']['topic']
- @must_be('user') + @need_right('edit') @need_extension('Flow') def moderate_topic(self, page, state, reason): """ @@ -7365,7 +7392,7 @@ data = req.submit() return data['flow']['moderate-topic']['committed']['topic']
- @must_be('user', 'flow-delete') + @need_right('flow-delete') @need_extension('Flow') def delete_topic(self, page, reason): """ @@ -7380,7 +7407,7 @@ """ return self.moderate_topic(page, 'delete', reason)
- @must_be('user', 'flow-hide') + @need_right('flow-hide') @need_extension('Flow') def hide_topic(self, page, reason): """ @@ -7395,7 +7422,7 @@ """ return self.moderate_topic(page, 'hide', reason)
- @must_be('user', 'flow-suppress') + @need_right('flow-suppress') @need_extension('Flow') def suppress_topic(self, page, reason): """ @@ -7410,7 +7437,7 @@ """ return self.moderate_topic(page, 'suppress', reason)
- @must_be('user') + @need_right('edit') @need_extension('Flow') def restore_topic(self, page, reason): """ @@ -7425,7 +7452,7 @@ """ return self.moderate_topic(page, 'restore', reason)
- @must_be('user') + @need_right('edit') @need_extension('Flow') def moderate_post(self, post, state, reason): """ @@ -7450,7 +7477,7 @@ data = req.submit() return data['flow']['moderate-post']['committed']['topic']
- @must_be('user', 'flow-delete') + @need_right('flow-delete') @need_extension('Flow') def delete_post(self, post, reason): """ @@ -7465,7 +7492,7 @@ """ return self.moderate_post(post, 'delete', reason)
- @must_be('user', 'flow-hide') + @need_right('flow-hide') @need_extension('Flow') def hide_post(self, post, reason): """ @@ -7480,7 +7507,7 @@ """ return self.moderate_post(post, 'hide', reason)
- @must_be('user', 'flow-suppress') + @need_right('flow-suppress') @need_extension('Flow') def suppress_post(self, post, reason): """ @@ -7495,7 +7522,7 @@ """ return self.moderate_post(post, 'suppress', reason)
- @must_be('user') + @need_right('edit') @need_extension('Flow') def restore_post(self, post, reason): """ @@ -7956,7 +7983,7 @@ return dtype
@deprecated_args(identification='entity') - @must_be(group='user') + @need_right('edit') def editEntity(self, entity, data, bot=True, **kwargs): """ Edit entity. @@ -8010,7 +8037,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def addClaim(self, entity, claim, bot=True, summary=None): """ Add a claim. @@ -8041,7 +8068,7 @@ entity.claims[claim.getID()] = [claim] entity.latest_revision_id = data['pageinfo']['lastrevid']
- @must_be(group='user') + @need_right('edit') def changeClaimTarget(self, claim, snaktype='value', bot=True, summary=None): """ @@ -8073,7 +8100,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def save_claim(self, claim, summary=None, bot=True): """ Save the whole claim to the wikibase site. @@ -8103,7 +8130,7 @@ claim.on_item.latest_revision_id = data['pageinfo']['lastrevid'] return data
- @must_be(group='user') + @need_right('edit') def editSource(self, claim, source, new=False, bot=True, summary=None, baserevid=None): """ @@ -8158,7 +8185,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def editQualifier(self, claim, qualifier, new=False, bot=True, summary=None, baserevid=None): """ @@ -8196,7 +8223,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def removeClaims(self, claims, bot=True, summary=None, baserevid=None): """ Remove claims. @@ -8230,7 +8257,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def removeSources(self, claim, sources, bot=True, summary=None, baserevid=None): """ @@ -8261,7 +8288,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def remove_qualifiers(self, claim, qualifiers, bot=True, summary=None, baserevid=None): """ @@ -8293,7 +8320,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def linkTitles(self, page1, page2, bot=True): """ Link two pages together. @@ -8321,7 +8348,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('item-merge') @deprecated_args(ignoreconflicts='ignore_conflicts', fromItem='from_item', toItem='to_item') def mergeItems(self, from_item, to_item, ignore_conflicts=None, @@ -8358,7 +8385,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('item-redirect') def set_redirect_target(self, from_item, to_item, bot=True): """ Make a redirect to another item. @@ -8381,7 +8408,7 @@ data = req.submit() return data
- @must_be(group='user') + @need_right('edit') def createNewItemFromPage(self, page, bot=True, **kwargs): """ Create a new Wikibase item for a provided page. diff --git a/tests/dry_site_tests.py b/tests/dry_site_tests.py index a37454f..5a4652f 100644 --- a/tests/dry_site_tests.py +++ b/tests/dry_site_tests.py @@ -9,7 +9,7 @@
import pywikibot from pywikibot.tools import deprecated -from pywikibot.site import must_be, need_version +from pywikibot.site import must_be, need_right, need_version from pywikibot.comms.http import user_agent from pywikibot.exceptions import UserRightsError
@@ -136,27 +136,38 @@ self.family.name = 'test' self.sitename = self.family.name + ':' + self.code self._logged_in_as = None + self._userinfo = [] self.obsolete = False super(TestMustBe, self).setUp() self.version = lambda: '1.14' # lowest supported release
- def login(self, sysop): - """Fake the log in and just store who logged in.""" - self._logged_in_as = 'sysop' if sysop else 'user' + def login(self, group): + """Fake the log in as required user group.""" + self._logged_in_as = group + self._userinfo = [group] + + def user(self): + """Fake the logged in user.""" + return self._logged_in_as
def has_group(self, group): """Fake the groups user belongs to.""" - return False + return group in self._userinfo
def testMockInTest(self): """Test that setUp and login work.""" self.assertIsNone(self._logged_in_as) - self.login(True) - self.assertEqual(self._logged_in_as, 'sysop') + self.login('user') + self.assertEqual(self._logged_in_as, 'user')
# Test that setUp is actually called between each test testMockInTestReset = testMockInTest # noqa: N815
+ @must_be('steward') + def call_this_steward_req_function(self, *args, **kwargs): + """Require a sysop to function.""" + return args, kwargs + @must_be('sysop') def call_this_sysop_req_function(self, *args, **kwargs): """Require a sysop to function.""" @@ -167,43 +178,125 @@ """Require a user to function.""" return args, kwargs
+ def testMustBeSteward(self): + """Test a function which requires a sysop.""" + args = (1, 2, 'a', 'b') + kwargs = {'i': 'j', 'k': 'l'} + self.login('steward') + retval = self.call_this_steward_req_function(*args, **kwargs) + self.assertEqual(retval[0], args) + self.assertEqual(retval[1], kwargs) + def testMustBeSysop(self): """Test a function which requires a sysop.""" args = (1, 2, 'a', 'b') kwargs = {'i': 'j', 'k': 'l'} + self.login('sysop') retval = self.call_this_sysop_req_function(*args, **kwargs) self.assertEqual(retval[0], args) self.assertEqual(retval[1], kwargs) - self.assertEqual(self._logged_in_as, 'sysop') + self.assertRaises(UserRightsError, self.call_this_steward_req_function, + args, kwargs)
def testMustBeUser(self): """Test a function which requires a user.""" args = (1, 2, 'a', 'b') kwargs = {'i': 'j', 'k': 'l'} + self.login('user') retval = self.call_this_user_req_function(*args, **kwargs) self.assertEqual(retval[0], args) self.assertEqual(retval[1], kwargs) - self.assertEqual(self._logged_in_as, 'user') + self.assertRaises(UserRightsError, self.call_this_sysop_req_function, + args, kwargs)
def testOverrideUserType(self): """Test overriding the required group.""" args = (1, 2, 'a', 'b') kwargs = {'i': 'j', 'k': 'l'} + self.login('sysop') retval = self.call_this_user_req_function( *args, as_group='sysop', **kwargs) self.assertEqual(retval[0], args) self.assertEqual(retval[1], kwargs) - self.assertEqual(self._logged_in_as, 'sysop')
def testObsoleteSite(self): """Test when the site is obsolete and shouldn't be edited.""" self.obsolete = True args = (1, 2, 'a', 'b') kwargs = {'i': 'j', 'k': 'l'} + self.login('steward') + retval = self.call_this_user_req_function(*args, **kwargs) + self.assertEqual(retval[0], args) + self.assertEqual(retval[1], kwargs) + self.login('user') self.assertRaises(UserRightsError, self.call_this_user_req_function, args, kwargs)
+class TestNeedRight(DebugOnlyTestCase): + + """Test cases for the must_be decorator.""" + + net = False + + # Implemented without setUpClass(cls) and global variables as objects + # were not completely disposed and recreated but retained 'memory' + def setUp(self): + """Creating fake variables to appear as a site.""" + self.code = 'test' + self.family = lambda: None + self.family.name = 'test' + self.sitename = self.family.name + ':' + self.code + self._logged_in_as = None + self._userinfo = [] + self.obsolete = False + super(TestNeedRight, self).setUp() + self.version = lambda: '1.14' # lowest supported release + + def login(self, group, right): + """Fake the log in as required user group.""" + self._logged_in_as = group + self._userinfo = [right] + + def user(self): + """Fake the logged in user.""" + return self._logged_in_as + + def has_right(self, right): + """Fake the groups user belongs to.""" + return right in self._userinfo + + @need_right('edit') + def call_this_edit_req_function(self, *args, **kwargs): + """Require a sysop to function.""" + return args, kwargs + + @need_right('move') + def call_this_move_req_function(self, *args, **kwargs): + """Require a sysop to function.""" + return args, kwargs + + def testNeedRightEdit(self): + """Test a function which requires a sysop.""" + args = (1, 2, 'a', 'b') + kwargs = {'i': 'j', 'k': 'l'} + self.login('user', 'edit') + retval = self.call_this_edit_req_function(*args, **kwargs) + self.assertEqual(retval[0], args) + self.assertEqual(retval[1], kwargs) + + def testNeedRightMove(self): + """Test a function which requires a sysop.""" + args = (1, 2, 'a', 'b') + kwargs = {'i': 'j', 'k': 'l'} + self.login('user', 'move') + retval = self.call_this_move_req_function(*args, **kwargs) + self.assertEqual(retval[0], args) + self.assertEqual(retval[1], kwargs) + self.assertRaises(UserRightsError, self.call_this_edit_req_function, + args, kwargs) + + class TestNeedVersion(DeprecationTestCase):
"""Test cases for the need_version decorator."""