[Pywikibot-commits] [Gerrit] ...core[master]: [IMPR] move TokenWallet to site/_tokenwallet.py file

jenkins-bot has submitted this change. ( https://gerrit.wikimedia.org/r/c/pywikibot/core/+/640399 ) Change subject: [IMPR] move TokenWallet to site/_tokenwallet.py file ...................................................................... [IMPR] move TokenWallet to site/_tokenwallet.py file Change-Id: I1c13ecad768ca8ca72493f3fcf906075bfce31d0 --- M docs/api_ref/pywikibot.site.rst M pywikibot/CONTENT.rst M pywikibot/site/__init__.py A pywikibot/site/_tokenwallet.py M tests/__init__.py M tests/site_tests.py A tests/token_tests.py 7 files changed, 328 insertions(+), 286 deletions(-) Approvals: Xqt: Looks good to me, approved jenkins-bot: Verified diff --git a/docs/api_ref/pywikibot.site.rst b/docs/api_ref/pywikibot.site.rst index 6e5796e..9545088 100644 --- a/docs/api_ref/pywikibot.site.rst +++ b/docs/api_ref/pywikibot.site.rst @@ -20,3 +20,8 @@ -------------------------------- .. automodule:: pywikibot.site._siteinfo + +pywikibot.site.\_tokenwallet module +----------------------------------- + +.. automodule:: pywikibot.site._tokenwallet diff --git a/pywikibot/CONTENT.rst b/pywikibot/CONTENT.rst index f34e7fe..fc9fa1e 100644 --- a/pywikibot/CONTENT.rst +++ b/pywikibot/CONTENT.rst @@ -119,6 +119,8 @@ +----------------------------+------------------------------------------------------+ | _siteinfo.py | Objects representing site info data contents. | +----------------------------+------------------------------------------------------+ + | _tokenwallet.py | Objects representing api tokens. | + +----------------------------+------------------------------------------------------+ +----------------------------+------------------------------------------------------+ diff --git a/pywikibot/site/__init__.py b/pywikibot/site/__init__.py index 9526291..3b47505 100644 --- a/pywikibot/site/__init__.py +++ b/pywikibot/site/__init__.py @@ -69,6 +69,7 @@ from pywikibot.site._decorators import need_extension, need_right, need_version from pywikibot.site._interwikimap import _InterwikiMap from pywikibot.site._siteinfo import Siteinfo +from pywikibot.site._tokenwallet import TokenWallet from pywikibot.throttle import Throttle from pywikibot.tools import ( ComparableMixin, @@ -1108,87 +1109,6 @@ return api.encode_url(query) -class TokenWallet: - - """Container for tokens.""" - - def __init__(self, site): - """Initializer. - - @type site: pywikibot.site.APISite - """ - self.site = site - self._tokens = {} - self.failed_cache = set() # cache unavailable tokens. - - def load_tokens(self, types, all=False): - """ - Preload one or multiple tokens. - - @param types: the types of token. - @type types: iterable - @param all: load all available tokens, if None only if it can be done - in one request. - @type all: bool - """ - if self.site.user() is None: - self.site.login() - - self._tokens.setdefault(self.site.user(), {}).update( - self.site.get_tokens(types, all=all)) - - # Preload all only the first time. - # When all=True types is extended in site.get_tokens(). - # Keys not recognised as tokens, are cached so they are not requested - # any longer. - if all is not False: - for key in types: - if key not in self._tokens[self.site.user()]: - self.failed_cache.add((self.site.user(), key)) - - def __getitem__(self, key): - """Get token value for the given key.""" - if self.site.user() is None: - self.site.login() - - user_tokens = self._tokens.setdefault(self.site.user(), {}) - # always preload all for users without tokens - failed_cache_key = (self.site.user(), key) - - try: - key = self.site.validate_tokens([key])[0] - except IndexError: - raise Error( - "Requested token '{0}' is invalid on {1} wiki." - .format(key, self.site)) - - if (key not in user_tokens - and failed_cache_key not in self.failed_cache): - self.load_tokens([key], all=False if user_tokens else None) - - if key in user_tokens: - return user_tokens[key] - else: - # token not allowed for self.site.user() on self.site - self.failed_cache.add(failed_cache_key) - # to be changed back to a plain KeyError? - raise Error( - "Action '{0}' is not allowed for user {1} on {2} wiki." - .format(key, self.site.user(), self.site)) - - def __contains__(self, key): - """Return True if the given token name is cached.""" - return key in self._tokens.setdefault(self.site.user(), {}) - - def __str__(self): - """Return a str representation of the internal tokens dictionary.""" - return self._tokens.__str__() - - def __repr__(self): - """Return a representation of the internal tokens dictionary.""" - return self._tokens.__repr__() - - class RemovedSite(BaseSite): """Site removed from a family.""" diff --git a/pywikibot/site/_tokenwallet.py b/pywikibot/site/_tokenwallet.py new file mode 100644 index 0000000..b161b29 --- /dev/null +++ b/pywikibot/site/_tokenwallet.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""Objects representing api tokens.""" +# +# (C) Pywikibot team, 2008-2020 +# +# Distributed under the terms of the MIT license. +# +from pywikibot.exceptions import Error + + +class TokenWallet: + + """Container for tokens.""" + + def __init__(self, site): + """Initializer. + + @type site: pywikibot.site.APISite + """ + self.site = site + self._tokens = {} + self.failed_cache = set() # cache unavailable tokens. + + def load_tokens(self, types, all=False): + """ + Preload one or multiple tokens. + + @param types: the types of token. + @type types: iterable + @param all: load all available tokens, if None only if it can be done + in one request. + @type all: bool + """ + if self.site.user() is None: + self.site.login() + + self._tokens.setdefault(self.site.user(), {}).update( + self.site.get_tokens(types, all=all)) + + # Preload all only the first time. + # When all=True types is extended in site.get_tokens(). + # Keys not recognised as tokens, are cached so they are not requested + # any longer. + if all is not False: + for key in types: + if key not in self._tokens[self.site.user()]: + self.failed_cache.add((self.site.user(), key)) + + def __getitem__(self, key): + """Get token value for the given key.""" + if self.site.user() is None: + self.site.login() + + user_tokens = self._tokens.setdefault(self.site.user(), {}) + # always preload all for users without tokens + failed_cache_key = (self.site.user(), key) + + try: + key = self.site.validate_tokens([key])[0] + except IndexError: + raise Error( + "Requested token '{0}' is invalid on {1} wiki." + .format(key, self.site)) + + if (key not in user_tokens + and failed_cache_key not in self.failed_cache): + self.load_tokens([key], all=False if user_tokens else None) + + if key in user_tokens: + return user_tokens[key] + else: + # token not allowed for self.site.user() on self.site + self.failed_cache.add(failed_cache_key) + # to be changed back to a plain KeyError? + raise Error( + "Action '{0}' is not allowed for user {1} on {2} wiki." + .format(key, self.site.user(), self.site)) + + def __contains__(self, key): + """Return True if the given token name is cached.""" + return key in self._tokens.setdefault(self.site.user(), {}) + + def __str__(self): + """Return a str representation of the internal tokens dictionary.""" + return self._tokens.__str__() + + def __repr__(self): + """Return a representation of the internal tokens dictionary.""" + return self._tokens.__repr__() diff --git a/tests/__init__.py b/tests/__init__.py index f313e60..e03eba8 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -114,6 +114,7 @@ 'timestamp', 'timestripper', 'tk', + 'token', 'tools', 'tools_chars', 'tools_formatter', diff --git a/tests/site_tests.py b/tests/site_tests.py index 0806f34..1d37ee8 100644 --- a/tests/site_tests.py +++ b/tests/site_tests.py @@ -29,7 +29,6 @@ DefaultWikidataClientTestCase, DeprecationTestCase, TestCase, - TestCaseBase, unittest, WikimediaDefaultSiteTestCase, WikidataTestCase, @@ -37,34 +36,6 @@ from tests.basepage import BasePageLoadRevisionsCachingTestBase -class TokenTestBase(TestCaseBase): - - """Verify token exists before running tests.""" - - def setUp(self): - """Skip test if user does not have token and clear site wallet.""" - super().setUp() - mysite = self.get_site() - ttype = self.token_type - try: - token = mysite.tokens[ttype] - except pywikibot.Error as error_msg: - self.assertRegex( - str(error_msg), - "Action '[a-z]+' is not allowed for user .* on .* wiki.") - self.assertNotIn(self.token_type, self.site.tokens) - self.skipTest(error_msg) - - self.token = token - self._orig_wallet = self.site.tokens - self.site.tokens = pywikibot.site.TokenWallet(self.site) - - def tearDown(self): - """Restore site tokens.""" - self.site.tokens = self._orig_wallet - super().tearDown() - - class TestSiteObjectDeprecatedFunctions(DefaultSiteTestCase, DeprecationTestCase): @@ -2099,59 +2070,6 @@ self.assertEqual(cnt, len(all_users), 'Some test usernames not found') -class PatrolTestCase(TokenTestBase, TestCase): - - """Test patrol method.""" - - family = 'wikipedia' - code = 'test' - - user = True - token_type = 'patrol' - write = True - - def test_patrol(self): - """Test the site.patrol() method.""" - mysite = self.get_site() - - rc = list(mysite.recentchanges(total=1)) - if not rc: - self.skipTest('no recent changes to patrol') - - rc = rc[0] - - # site.patrol() needs params - self.assertRaises(pywikibot.Error, lambda x: list(x), mysite.patrol()) - try: - result = list(mysite.patrol(rcid=rc['rcid'])) - except api.APIError as error: - if error.code == 'permissiondenied': - self.skipTest(error) - raise - - if hasattr(mysite, '_patroldisabled') and mysite._patroldisabled: - self.skipTest('Patrolling is disabled on {} wiki.'.format(mysite)) - - result = result[0] - self.assertIsInstance(result, dict) - - params = {'rcid': 0} - if mysite.mw_version >= '1.22': - params['revid'] = [0, 1] - - raised = False - try: - # no such rcid, revid or too old revid - list(mysite.patrol(**params)) - except api.APIError as error: - if error.code == 'badtoken': - self.skipTest(error) - except pywikibot.Error: - # expected result - raised = True - self.assertTrue(raised, msg='pywikibot.Error not raised') - - class SiteRandomTestCase(DefaultSiteTestCase): """Test random methods of a site.""" @@ -2203,129 +2121,6 @@ self.assertIn(rndpage.namespace(), [6, 7]) -class TestSiteTokens(DefaultSiteTestCase): - - """Test cases for tokens in Site methods. - - Versions of sites are simulated if actual versions are higher than - needed by the test case. - - Test is skipped if site version is not compatible. - - """ - - user = True - - def setUp(self): - """Store version.""" - super().setUp() - self.mysite = self.get_site() - self._version = self.mysite.mw_version - self.orig_version = self.mysite.version - - def tearDown(self): - """Restore version.""" - super().tearDown() - self.mysite.version = self.orig_version - - def _test_tokens(self, version, test_version, additional_token): - """Test tokens.""" - if version and (self._version < version - or self._version < test_version): - raise unittest.SkipTest( - 'Site {} version {} is too low for this tests.' - .format(self.mysite, self._version)) - - self.mysite.version = lambda: test_version - - for ttype in ('edit', 'move', additional_token): - tokentype = self.mysite.validate_tokens([ttype]) - try: - token = self.mysite.tokens[ttype] - except pywikibot.Error as error_msg: - if tokentype: - self.assertRegex( - str(error_msg), - "Action '[a-z]+' is not allowed " - 'for user .* on .* wiki.') - # test __contains__ - self.assertNotIn(tokentype[0], self.mysite.tokens) - else: - self.assertRegex( - str(error_msg), - "Requested token '[a-z]+' is invalid on .* wiki.") - else: - self.assertIsInstance(token, str) - self.assertEqual(token, self.mysite.tokens[ttype]) - # test __contains__ - self.assertIn(tokentype[0], self.mysite.tokens) - - def test_tokens_in_mw_119(self): - """Test ability to get page tokens.""" - self._test_tokens(None, '1.19', 'delete') - - def test_patrol_tokens_in_mw_119(self): - """Test ability to get patrol token on MW 1.19 wiki.""" - self._test_tokens('1.19', '1.19', 'patrol') - - def test_tokens_in_mw_120_124wmf18(self): - """Test ability to get page tokens.""" - self._test_tokens('1.20', '1.21', 'deleteglobalaccount') - - def test_patrol_tokens_in_mw_120(self): - """Test ability to get patrol token.""" - self._test_tokens('1.19', '1.20', 'patrol') - - def test_tokens_in_mw_124wmf19(self): - """Test ability to get page tokens.""" - self._test_tokens('1.24wmf19', '1.24wmf20', 'deleteglobalaccount') - - def testInvalidToken(self): - """Test invalid token.""" - self.assertRaises(pywikibot.Error, lambda t: self.mysite.tokens[t], - 'invalidtype') - - -class TestDeprecatedEditTokenFunctions(TokenTestBase, - DefaultSiteTestCase, - DeprecationTestCase): - - """Test cases for Site edit token deprecated methods.""" - - cached = True - user = True - token_type = 'edit' - - def test_getToken(self): - """Test ability to get page tokens using site.getToken.""" - self.mysite = self.site - self.assertEqual(self.mysite.getToken(), self.mysite.tokens['edit']) - self.assertOneDeprecationParts('pywikibot.site.APISite.getToken', - "the 'tokens' property") - - -class TestDeprecatedPatrolToken(DefaultSiteTestCase, DeprecationTestCase): - - """Test cases for Site patrol token deprecated methods.""" - - cached = True - user = True - - def test_getPatrolToken(self): - """Test site.getPatrolToken.""" - self.mysite = self.site - try: - self.assertEqual(self.mysite.getPatrolToken(), - self.mysite.tokens['patrol']) - self.assertOneDeprecation() - except pywikibot.Error as error_msg: - self.assertRegex( - str(error_msg), - "Action '[a-z]+' is not allowed for user .* on .* wiki.") - # test __contains__ - self.assertNotIn('patrol', self.mysite.tokens) - - class TestSiteExtensions(WikimediaDefaultSiteTestCase): """Test cases for Site extensions.""" diff --git a/tests/token_tests.py b/tests/token_tests.py new file mode 100644 index 0000000..6e5ae57 --- /dev/null +++ b/tests/token_tests.py @@ -0,0 +1,230 @@ +# -*- coding: utf-8 -*- +"""Tests for tokens.""" +# +# (C) Pywikibot team, 2015-2020 +# +# Distributed under the terms of the MIT license. +# +from contextlib import suppress + +import pywikibot + +from pywikibot.data import api +from pywikibot.site import TokenWallet + +from tests.aspects import ( + DefaultSiteTestCase, + DeprecationTestCase, + TestCase, + TestCaseBase, + unittest, +) + + +class TestSiteTokens(DefaultSiteTestCase): + + """Test cases for tokens in Site methods. + + Versions of sites are simulated if actual versions are higher than + needed by the test case. + + Test is skipped if site version is not compatible. + + """ + + user = True + + def setUp(self): + """Store version.""" + super().setUp() + self.mysite = self.get_site() + self._version = self.mysite.mw_version + self.orig_version = self.mysite.version + + def tearDown(self): + """Restore version.""" + super().tearDown() + self.mysite.version = self.orig_version + + def _test_tokens(self, version, test_version, additional_token): + """Test tokens.""" + if version and (self._version < version + or self._version < test_version): + raise unittest.SkipTest( + 'Site {} version {} is too low for this tests.' + .format(self.mysite, self._version)) + + self.mysite.version = lambda: test_version + + for ttype in ('edit', 'move', additional_token): + tokentype = self.mysite.validate_tokens([ttype]) + try: + token = self.mysite.tokens[ttype] + except pywikibot.Error as error_msg: + if tokentype: + self.assertRegex( + str(error_msg), + "Action '[a-z]+' is not allowed " + 'for user .* on .* wiki.') + # test __contains__ + self.assertNotIn(tokentype[0], self.mysite.tokens) + else: + self.assertRegex( + str(error_msg), + "Requested token '[a-z]+' is invalid on .* wiki.") + else: + self.assertIsInstance(token, str) + self.assertEqual(token, self.mysite.tokens[ttype]) + # test __contains__ + self.assertIn(tokentype[0], self.mysite.tokens) + + def test_tokens_in_mw_119(self): + """Test ability to get page tokens.""" + self._test_tokens(None, '1.19', 'delete') + + def test_patrol_tokens_in_mw_119(self): + """Test ability to get patrol token on MW 1.19 wiki.""" + self._test_tokens('1.19', '1.19', 'patrol') + + def test_tokens_in_mw_120_124wmf18(self): + """Test ability to get page tokens.""" + self._test_tokens('1.20', '1.21', 'deleteglobalaccount') + + def test_patrol_tokens_in_mw_120(self): + """Test ability to get patrol token.""" + self._test_tokens('1.19', '1.20', 'patrol') + + def test_tokens_in_mw_124wmf19(self): + """Test ability to get page tokens.""" + self._test_tokens('1.24wmf19', '1.24wmf20', 'deleteglobalaccount') + + def test_invalid_token(self): + """Test invalid token.""" + self.assertRaises(pywikibot.Error, lambda t: self.mysite.tokens[t], + 'invalidtype') + + +class TokenTestBase(TestCaseBase): + + """Verify token exists before running tests.""" + + def setUp(self): + """Skip test if user does not have token and clear site wallet.""" + super().setUp() + mysite = self.get_site() + ttype = self.token_type + try: + token = mysite.tokens[ttype] + except pywikibot.Error as error_msg: + self.assertRegex( + str(error_msg), + "Action '[a-z]+' is not allowed for user .* on .* wiki.") + self.assertNotIn(self.token_type, self.site.tokens) + self.skipTest(error_msg) + + self.token = token + self._orig_wallet = self.site.tokens + self.site.tokens = TokenWallet(self.site) + + def tearDown(self): + """Restore site tokens.""" + self.site.tokens = self._orig_wallet + super().tearDown() + + +class PatrolTestCase(TokenTestBase, TestCase): + + """Test patrol method.""" + + family = 'wikipedia' + code = 'test' + + user = True + token_type = 'patrol' + write = True + + def test_patrol(self): + """Test the site.patrol() method.""" + mysite = self.get_site() + + rc = list(mysite.recentchanges(total=1)) + if not rc: + self.skipTest('no recent changes to patrol') + + rc = rc[0] + + # site.patrol() needs params + self.assertRaises(pywikibot.Error, lambda x: list(x), mysite.patrol()) + try: + result = list(mysite.patrol(rcid=rc['rcid'])) + except api.APIError as error: + if error.code == 'permissiondenied': + self.skipTest(error) + raise + + if hasattr(mysite, '_patroldisabled') and mysite._patroldisabled: + self.skipTest('Patrolling is disabled on {} wiki.'.format(mysite)) + + result = result[0] + self.assertIsInstance(result, dict) + + params = {'rcid': 0} + if mysite.mw_version >= '1.22': + params['revid'] = [0, 1] + + raised = False + try: + # no such rcid, revid or too old revid + list(mysite.patrol(**params)) + except api.APIError as error: + if error.code == 'badtoken': + self.skipTest(error) + except pywikibot.Error: + # expected result + raised = True + self.assertTrue(raised, msg='pywikibot.Error not raised') + + +class TestDeprecatedEditTokenFunctions(TokenTestBase, + DefaultSiteTestCase, + DeprecationTestCase): + + """Test cases for Site edit token deprecated methods.""" + + cached = True + user = True + token_type = 'edit' + + def test_get_token(self): + """Test ability to get page tokens using site.getToken.""" + self.mysite = self.site + self.assertEqual(self.mysite.getToken(), self.mysite.tokens['edit']) + self.assertOneDeprecationParts('pywikibot.site.APISite.getToken', + "the 'tokens' property") + + +class TestDeprecatedPatrolToken(DefaultSiteTestCase, DeprecationTestCase): + + """Test cases for Site patrol token deprecated methods.""" + + cached = True + user = True + + def test_get_patrol_token(self): + """Test site.getPatrolToken.""" + self.mysite = self.site + try: + self.assertEqual(self.mysite.getPatrolToken(), + self.mysite.tokens['patrol']) + self.assertOneDeprecation() + except pywikibot.Error as error_msg: + self.assertRegex( + str(error_msg), + "Action '[a-z]+' is not allowed for user .* on .* wiki.") + # test __contains__ + self.assertNotIn('patrol', self.mysite.tokens) + + +if __name__ == '__main__': # pragma: no cover + with suppress(SystemExit): + unittest.main() -- To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/640399 To unsubscribe, or for help writing mail filters, visit https://gerrit.wikimedia.org/r/settings Gerrit-Project: pywikibot/core Gerrit-Branch: master Gerrit-Change-Id: I1c13ecad768ca8ca72493f3fcf906075bfce31d0 Gerrit-Change-Number: 640399 Gerrit-PatchSet: 2 Gerrit-Owner: Xqt &lt;info(a)gno.de&gt; Gerrit-Reviewer: Xqt &lt;info(a)gno.de&gt; Gerrit-Reviewer: jenkins-bot Gerrit-MessageType: merged