import json import logging import sys import time import unittest import unittest.mock as mock from copy import deepcopy from datetime import date from io import StringIO import mwclient import pkg_resources # part of setuptools import pytest import requests import responses from requests_oauthlib import OAuth1 if __name__ == "__main__": print() print("Note: Running in stand-alone mode. Consult the README") print(" (section 'Contributing') for advice on running tests.") print() logging.basicConfig(level=logging.DEBUG) class TestCase(unittest.TestCase): def metaResponse(self, **kwargs): tpl = '{"query":{"general":{"generator":"MediaWiki %(version)s"},"namespaces":{"-1":{"*":"Special","canonical":"Special","case":"first-letter","id":-1},"-2":{"*":"Media","canonical":"Media","case":"first-letter","id":-2},"0":{"*":"","case":"first-letter","content":"","id":0},"1":{"*":"Talk","canonical":"Talk","case":"first-letter","id":1,"subpages":""},"10":{"*":"Template","canonical":"Template","case":"first-letter","id":10,"subpages":""},"100":{"*":"Test namespace 1","canonical":"Test namespace 1","case":"first-letter","id":100,"subpages":""},"101":{"*":"Test namespace 1 talk","canonical":"Test namespace 1 talk","case":"first-letter","id":101,"subpages":""},"102":{"*":"Test namespace 2","canonical":"Test namespace 2","case":"first-letter","id":102,"subpages":""},"103":{"*":"Test namespace 2 talk","canonical":"Test namespace 2 talk","case":"first-letter","id":103,"subpages":""},"11":{"*":"Template talk","canonical":"Template talk","case":"first-letter","id":11,"subpages":""},"1198":{"*":"Translations","canonical":"Translations","case":"first-letter","id":1198,"subpages":""},"1199":{"*":"Translations talk","canonical":"Translations talk","case":"first-letter","id":1199,"subpages":""},"12":{"*":"Help","canonical":"Help","case":"first-letter","id":12,"subpages":""},"13":{"*":"Help talk","canonical":"Help talk","case":"first-letter","id":13,"subpages":""},"14":{"*":"Category","canonical":"Category","case":"first-letter","id":14},"15":{"*":"Category talk","canonical":"Category talk","case":"first-letter","id":15,"subpages":""},"2":{"*":"User","canonical":"User","case":"first-letter","id":2,"subpages":""},"2500":{"*":"VisualEditor","canonical":"VisualEditor","case":"first-letter","id":2500},"2501":{"*":"VisualEditor talk","canonical":"VisualEditor talk","case":"first-letter","id":2501},"2600":{"*":"Topic","canonical":"Topic","case":"first-letter","defaultcontentmodel":"flow-board","id":2600},"3":{"*":"User talk","canonical":"User talk","case":"first-letter","id":3,"subpages":""},"4":{"*":"Wikipedia","canonical":"Project","case":"first-letter","id":4,"subpages":""},"460":{"*":"Campaign","canonical":"Campaign","case":"case-sensitive","defaultcontentmodel":"Campaign","id":460},"461":{"*":"Campaign talk","canonical":"Campaign talk","case":"case-sensitive","id":461},"5":{"*":"Wikipedia talk","canonical":"Project talk","case":"first-letter","id":5,"subpages":""},"6":{"*":"File","canonical":"File","case":"first-letter","id":6},"7":{"*":"File talk","canonical":"File talk","case":"first-letter","id":7,"subpages":""},"710":{"*":"TimedText","canonical":"TimedText","case":"first-letter","id":710},"711":{"*":"TimedText talk","canonical":"TimedText talk","case":"first-letter","id":711},"8":{"*":"MediaWiki","canonical":"MediaWiki","case":"first-letter","id":8,"subpages":""},"828":{"*":"Module","canonical":"Module","case":"first-letter","id":828,"subpages":""},"829":{"*":"Module talk","canonical":"Module talk","case":"first-letter","id":829,"subpages":""},"866":{"*":"CNBanner","canonical":"CNBanner","case":"first-letter","id":866},"867":{"*":"CNBanner talk","canonical":"CNBanner talk","case":"first-letter","id":867,"subpages":""},"9":{"*":"MediaWiki talk","canonical":"MediaWiki talk","case":"first-letter","id":9,"subpages":""},"90":{"*":"Thread","canonical":"Thread","case":"first-letter","id":90},"91":{"*":"Thread talk","canonical":"Thread talk","case":"first-letter","id":91},"92":{"*":"Summary","canonical":"Summary","case":"first-letter","id":92},"93":{"*":"Summary talk","canonical":"Summary talk","case":"first-letter","id":93}},"userinfo":{"anon":"","groups":["*"],"id":0,"name":"127.0.0.1","rights": %(rights)s}}}' tpl = tpl % {'version': kwargs.get('version', '1.24wmf17'), 'rights': json.dumps(kwargs.get('rights', ["createaccount", "read", "edit", "createpage", "createtalk", "writeapi", "editmyusercss", "editmyuserjs", "viewmywatchlist", "editmywatchlist", "viewmyprivateinfo", "editmyprivateinfo", "editmyoptions", "centralauth-merge", "abusefilter-view", "abusefilter-log", "translate", "vipsscaler-test", "upload"])) } res = json.loads(tpl) if kwargs.get('writeapi', True): res['query']['general']['writeapi'] = '' return res def metaResponseAsJson(self, **kwargs): return json.dumps(self.metaResponse(**kwargs)) def httpShouldReturn(self, body=None, callback=None, scheme='https', host='test.wikipedia.org', path='/w/', script='api', headers=None, status=200, method='GET'): url = '{scheme}://{host}{path}{script}.php'.format(scheme=scheme, host=host, path=path, script=script) mock = responses.GET if method == 'GET' else responses.POST if body is None: responses.add_callback(mock, url, callback=callback) else: responses.add(mock, url, body=body, content_type='application/json', adding_headers=headers, status=status) def stdSetup(self): self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org') responses.reset() return site def makePageResponse(self, title='Dummy.jpg', **kwargs): # Creates a dummy page response pageinfo = { "contentmodel": "wikitext", "lastrevid": 112353797, "length": 389, "ns": 6, "pageid": 738154, "pagelanguage": "en", "protection": [], "title": title, "touched": "2014-09-10T20:37:25Z" } pageinfo.update(**kwargs) res = { "query": { "pages": { "9": pageinfo } } } return json.dumps(res) class TestClient(TestCase): def setUp(self): pass def testVersion(self): # The version specified in setup.py should equal the one specified in client.py version = pkg_resources.require("mwclient")[0].version assert version == mwclient.__version__ @responses.activate def test_https_as_default(self): # 'https' should be the default scheme self.httpShouldReturn(self.metaResponseAsJson(), scheme='https') site = mwclient.Site('test.wikipedia.org') assert len(responses.calls) == 1 assert responses.calls[0].request.method == 'GET' @responses.activate def test_max_lag(self): # Client should wait and retry if lag exceeds max-lag def request_callback(request): if len(responses.calls) == 0: return (200, {'x-database-lag': '0', 'retry-after': '0'}, '') else: return (200, {}, self.metaResponseAsJson()) self.httpShouldReturn(callback=request_callback, scheme='https') site = mwclient.Site('test.wikipedia.org') assert len(responses.calls) == 2 assert 'retry-after' in responses.calls[0].response.headers assert 'retry-after' not in responses.calls[1].response.headers @responses.activate def test_http_error(self): # Client should raise HTTPError self.httpShouldReturn('Uh oh', scheme='https', status=400) with pytest.raises(requests.exceptions.HTTPError): site = mwclient.Site('test.wikipedia.org') @responses.activate def test_force_http(self): # Setting http should work self.httpShouldReturn(self.metaResponseAsJson(), scheme='http') site = mwclient.Site('test.wikipedia.org', scheme='http') assert len(responses.calls) == 1 @responses.activate def test_user_agent_is_sent(self): # User specified user agent should be sent sent to server self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org', clients_useragent='MyFabulousClient') assert 'MyFabulousClient' in responses.calls[0].request.headers['user-agent'] @responses.activate def test_custom_headers_are_sent(self): # Custom headers should be sent to the server self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org', custom_headers={'X-Wikimedia-Debug': 'host=mw1099.eqiad.wmnet; log'}) assert 'host=mw1099.eqiad.wmnet; log' in responses.calls[0].request.headers['X-Wikimedia-Debug'] @responses.activate def test_basic_request(self): self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org') assert 'action=query' in responses.calls[0].request.url assert 'meta=siteinfo%7Cuserinfo' in responses.calls[0].request.url @responses.activate def test_httpauth_defaults_to_basic_auth(self): self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org', httpauth=('me', 'verysecret')) assert isinstance(site.connection.auth, requests.auth.HTTPBasicAuth) @responses.activate def test_basic_auth_non_latin(self): self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org', httpauth=('我', '非常秘密')) assert isinstance(site.connection.auth, requests.auth.HTTPBasicAuth) @responses.activate def test_httpauth_raise_error_on_invalid_type(self): self.httpShouldReturn(self.metaResponseAsJson()) with pytest.raises(RuntimeError): site = mwclient.Site('test.wikipedia.org', httpauth=1) @responses.activate def test_oauth(self): self.httpShouldReturn(self.metaResponseAsJson()) site = mwclient.Site('test.wikipedia.org', consumer_token='a', consumer_secret='b', access_token='c', access_secret='d') assert isinstance(site.connection.auth, OAuth1) @responses.activate def test_api_disabled(self): # Should raise APIDisabledError if API is not enabled self.httpShouldReturn('MediaWiki API is not enabled for this site.') with pytest.raises(mwclient.errors.APIDisabledError): site = mwclient.Site('test.wikipedia.org') @responses.activate def test_version(self): # Should parse the MediaWiki version number correctly self.httpShouldReturn(self.metaResponseAsJson(version='1.16')) site = mwclient.Site('test.wikipedia.org') assert site.initialized is True assert site.version == (1, 16) @responses.activate def test_min_version(self): # Should raise MediaWikiVersionError if API version is < 1.16 self.httpShouldReturn(self.metaResponseAsJson(version='1.15')) with pytest.raises(mwclient.errors.MediaWikiVersionError): site = mwclient.Site('test.wikipedia.org') @responses.activate def test_private_wiki(self): # Should not raise error self.httpShouldReturn(json.dumps({ 'error': { 'code': 'readapidenied', 'info': 'You need read permission to use this module' } })) site = mwclient.Site('test.wikipedia.org') assert site.initialized is False # ----- Use standard setup for rest @responses.activate def test_headers(self): # Content-type should be 'application/x-www-form-urlencoded' for POST requests site = self.stdSetup() self.httpShouldReturn('{}', method='POST') site.post('purge', title='Main Page') assert len(responses.calls) == 1 assert 'content-type' in responses.calls[0].request.headers assert responses.calls[0].request.headers['content-type'] == 'application/x-www-form-urlencoded' @responses.activate def test_raw_index(self): # Initializing the client should result in one request site = self.stdSetup() self.httpShouldReturn('Some data', script='index') site.raw_index(action='purge', title='Main Page', http_method='GET') assert len(responses.calls) == 1 @responses.activate def test_api_error_response(self): # Test that APIError is thrown on error response site = self.stdSetup() self.httpShouldReturn(json.dumps({ 'error': { 'code': 'assertuserfailed', 'info': 'Assertion that the user is logged in failed', '*': 'See https://en.wikipedia.org/w/api.php for API usage' } }), method='POST') with pytest.raises(mwclient.errors.APIError) as excinfo: site.api(action='edit', title='Wikipedia:Sandbox') assert excinfo.value.code == 'assertuserfailed' assert excinfo.value.info == 'Assertion that the user is logged in failed' assert len(responses.calls) == 1 @responses.activate def test_smw_error_response(self): # Test that APIError is thrown on error response from SMW site = self.stdSetup() self.httpShouldReturn(json.dumps({ 'error': { 'query': 'Certains « <nowiki>[[</nowiki> » dans votre requête n’ont pas été clos par des « ]] » correspondants.' } }), method='GET') with pytest.raises(mwclient.errors.APIError) as excinfo: list(site.ask('test')) assert excinfo.value.code is None assert excinfo.value.info == 'Certains « <nowiki>[[</nowiki> » dans votre requête n’ont pas été clos par des « ]] » correspondants.' assert len(responses.calls) == 1 @responses.activate def test_smw_response_v0_5(self): # Test that the old SMW results format is handled site = self.stdSetup() self.httpShouldReturn(json.dumps({ "query": { "results": [ { "exists": "", "fulltext": "Indeks (bibliotekfag)", "fullurl": "http://example.com/wiki/Indeks_(bibliotekfag)", "namespace": 0, "printouts": [ { "0": "1508611329", "label": "Endringsdato" } ] }, { "exists": "", "fulltext": "Serendipitet", "fullurl": "http://example.com/wiki/Serendipitet", "namespace": 0, "printouts": [ { "0": "1508611394", "label": "Endringsdato" } ] } ], "serializer": "SMW\\Serializers\\QueryResultSerializer", "version": 0.5 } }), method='GET') answers = set(result['fulltext'] for result in site.ask('test')) assert answers == set(('Serendipitet', 'Indeks (bibliotekfag)')) @responses.activate def test_smw_response_v2(self): # Test that the new SMW results format is handled site = self.stdSetup() self.httpShouldReturn(json.dumps({ "query": { "results": { "Indeks (bibliotekfag)": { "exists": "1", "fulltext": "Indeks (bibliotekfag)", "fullurl": "http://example.com/wiki/Indeks_(bibliotekfag)", "namespace": 0, "printouts": { "Endringsdato": [{ "raw": "1/2017/10/17/22/50/4/0", "label": "Endringsdato" }] } }, "Serendipitet": { "exists": "1", "fulltext": "Serendipitet", "fullurl": "http://example.com/wiki/Serendipitet", "namespace": 0, "printouts": { "Endringsdato": [{ "raw": "1/2017/10/17/22/50/4/0", "label": "Endringsdato" }] } } }, "serializer": "SMW\\Serializers\\QueryResultSerializer", "version": 2 } }), method='GET') answers = set(result['fulltext'] for result in site.ask('test')) assert answers == set(('Serendipitet', 'Indeks (bibliotekfag)')) @responses.activate def test_repr(self): # Test repr() site = self.stdSetup() assert repr(site) == '<Site object \'test.wikipedia.org/w/\'>' @mock.patch("time.sleep") @responses.activate def test_api_http_error(self, timesleep): # Test error paths in raw_call, via raw_api as it's more # convenient to call. This would be way nicer with pytest # parametrization but you can't use parametrization inside # unittest.TestCase :( site = self.stdSetup() # All HTTP errors should raise HTTPError with or without retries self.httpShouldReturn(body="foo", status=401) with pytest.raises(requests.exceptions.HTTPError): site.raw_api("query", "GET") # for a 4xx response we should *not* retry assert timesleep.call_count == 0 with pytest.raises(requests.exceptions.HTTPError): site.raw_api("query", "GET", retry_on_error=False) self.httpShouldReturn(body="foo", status=503) with pytest.raises(requests.exceptions.HTTPError): site.raw_api("query", "GET") # for a 5xx response we *should* retry assert timesleep.call_count == 25 timesleep.reset_mock() with pytest.raises(requests.exceptions.HTTPError): site.raw_api("query", "GET", retry_on_error=False) # check we did not retry assert timesleep.call_count == 0 # stop sending bad statuses self.httpShouldReturn(body="foo", status=200) # ConnectionError should retry then pass through, takes # advantage of responses raising ConnectionError if you # hit a URL that hasn't been configured. Timeout follows # the same path so we don't bother testing it separately realhost = site.host site.host = "notthere" with pytest.raises(requests.exceptions.ConnectionError): site.raw_api("query", "GET") assert timesleep.call_count == 25 timesleep.reset_mock() with pytest.raises(requests.exceptions.ConnectionError): site.raw_api("query", "GET", retry_on_error=False) # check we did not retry assert timesleep.call_count == 0 @mock.patch("time.sleep") @responses.activate def test_api_dblag(self, timesleep): site = self.stdSetup() # db lag should retry then raise MaximumRetriesExceeded, # even with retry_on_error set self.httpShouldReturn( body="foo", status=200, headers={"x-database-lag": "true", "retry-after": "5"} ) with pytest.raises(mwclient.errors.MaximumRetriesExceeded): site.raw_api("query", "GET") assert timesleep.call_count == 25 timesleep.reset_mock() with pytest.raises(mwclient.errors.MaximumRetriesExceeded): site.raw_api("query", "GET", retry_on_error=False) assert timesleep.call_count == 25 class TestLogin(TestCase): @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_old_login_flow(self, raw_api, site_init): # The login flow used before MW 1.27 that starts with a action=login POST request login_token = 'abc+\\' def side_effect(*args, **kwargs): if 'lgtoken' not in kwargs: return { 'login': {'result': 'NeedToken', 'token': login_token} } elif 'lgname' in kwargs: assert kwargs['lgtoken'] == login_token return { 'login': {'result': 'Success'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.login('myusername', 'mypassword') call_args = raw_api.call_args_list assert len(call_args) == 3 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='login') assert call_args[1] == mock.call('login', 'POST', lgname='myusername', lgpassword='mypassword') assert call_args[2] == mock.call('login', 'POST', lgname='myusername', lgpassword='mypassword', lgtoken=login_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_new_login_flow(self, raw_api, site_init): # The login flow used from MW 1.27 that starts with a meta=tokens GET request login_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'logintoken': login_token}} } elif 'lgname' in kwargs: assert kwargs['lgtoken'] == login_token return { 'login': {'result': 'Success'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.login('myusername', 'mypassword') call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='login') assert call_args[1] == mock.call('login', 'POST', lgname='myusername', lgpassword='mypassword', lgtoken=login_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_clientlogin_success(self, raw_api, site_init): login_token = 'abc+\\' def api_side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'logintoken': login_token}} } elif 'username' in kwargs: assert kwargs['logintoken'] == login_token assert kwargs.get('loginreturnurl') return { 'clientlogin': {'status': 'PASS'} } raw_api.side_effect = api_side_effect site = mwclient.Site('test.wikipedia.org') # this would be done by site_init usually, but we're mocking it site.version = (1, 28, 0) success = site.clientlogin(username='myusername', password='mypassword') url = '%s://%s' % (site.scheme, site.host) call_args = raw_api.call_args_list assert success is True assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='login') assert call_args[1] == mock.call( 'clientlogin', 'POST', username='myusername', password='mypassword', loginreturnurl=url, logintoken=login_token ) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_clientlogin_fail(self, raw_api, site_init): login_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'logintoken': login_token}} } elif 'username' in kwargs: assert kwargs['logintoken'] == login_token assert kwargs.get('loginreturnurl') return { 'clientlogin': {'status': 'FAIL'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') # this would be done by site_init usually, but we're mocking it site.version = (1, 28, 0) with pytest.raises(mwclient.errors.LoginError): success = site.clientlogin(username='myusername', password='mypassword') call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='login') assert call_args[1] == mock.call( 'clientlogin', 'POST', username='myusername', password='mypassword', loginreturnurl='%s://%s' % (site.scheme, site.host), logintoken=login_token ) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_clientlogin_continue(self, raw_api, site_init): login_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'logintoken': login_token}} } elif 'username' in kwargs: assert kwargs['logintoken'] == login_token assert kwargs.get('loginreturnurl') return { 'clientlogin': {'status': 'UI'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') # this would be done by site_init usually, but we're mocking it site.version = (1, 28, 0) success = site.clientlogin(username='myusername', password='mypassword') url = '%s://%s' % (site.scheme, site.host) call_args = raw_api.call_args_list assert success == {'status': 'UI'} assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='login') assert call_args[1] == mock.call( 'clientlogin', 'POST', username='myusername', password='mypassword', loginreturnurl=url, logintoken=login_token ) class TestClientApiMethods(TestCase): def setUp(self): self.api = mock.patch('mwclient.client.Site.api').start() self.api.return_value = self.metaResponse() self.site = mwclient.Site('test.wikipedia.org') def tearDown(self): mock.patch.stopall() def test_revisions(self): self.api.return_value = { 'query': {'pages': {'1': { 'pageid': 1, 'title': 'Test page', 'revisions': [{ 'revid': 689697696, 'timestamp': '2015-11-08T21:52:46Z', 'comment': 'Test comment 1' }, { 'revid': 689816909, 'timestamp': '2015-11-09T16:09:28Z', 'comment': 'Test comment 2' }] }}}} revisions = [rev for rev in self.site.revisions([689697696, 689816909], prop='content')] args, kwargs = self.api.call_args assert kwargs.get('revids') == '689697696|689816909' assert len(revisions) == 2 assert revisions[0]['pageid'] == 1 assert revisions[0]['pagetitle'] == 'Test page' assert revisions[0]['revid'] == 689697696 assert revisions[0]['timestamp'] == time.strptime('2015-11-08T21:52:46Z', '%Y-%m-%dT%H:%M:%SZ') assert revisions[1]['revid'] == 689816909 class TestClientUploadArgs(TestCase): def setUp(self): self.raw_call = mock.patch('mwclient.client.Site.raw_call').start() def configure(self, rights=['read', 'upload']): self.raw_call.side_effect = [self.metaResponseAsJson(rights=rights)] self.site = mwclient.Site('test.wikipedia.org') self.vars = { 'fname': 'Some "ßeta" æøå.jpg', 'comment': 'Some slightly complex comment<br> π ≈ 3, © Me.jpg', 'token': 'abc+\\' } self.raw_call.side_effect = [ # 1st response: self.makePageResponse(title='File:Test.jpg', imagerepository='local', imageinfo=[{ "comment": "", "height": 1440, "metadata": [], "sha1": "69a764a9cf8307ea4130831a0aa0b9b7f9585726", "size": 123, "timestamp": "2013-12-22T07:11:07Z", "user": "TestUser", "width": 2160 }]), # 2nd response: json.dumps({'query': {'tokens': {'csrftoken': self.vars['token']}}}), # 3rd response: json.dumps({ "upload": { "result": "Success", "filename": self.vars['fname'], "imageinfo": [] } }) ] def tearDown(self): mock.patch.stopall() def test_upload_args(self): # Test that methods are called, and arguments sent as expected self.configure() self.site.upload(file=StringIO('test'), filename=self.vars['fname'], comment=self.vars['comment']) args, kwargs = self.raw_call.call_args data = args[1] files = args[2] assert data.get('action') == 'upload' assert data.get('filename') == self.vars['fname'] assert data.get('comment') == self.vars['comment'] assert data.get('token') == self.vars['token'] assert 'file' in files def test_upload_missing_filename(self): self.configure() with pytest.raises(TypeError): self.site.upload(file=StringIO('test')) def test_upload_ambigitious_args(self): self.configure() with pytest.raises(TypeError): self.site.upload(filename='Test', file=StringIO('test'), filekey='abc') def test_upload_missing_upload_permission(self): self.configure(rights=['read']) with pytest.raises(mwclient.errors.InsufficientPermission): self.site.upload(filename='Test', file=StringIO('test')) class TestClientGetTokens(TestCase): def setUp(self): self.raw_call = mock.patch('mwclient.client.Site.raw_call').start() def configure(self, version='1.24'): self.raw_call.return_value = self.metaResponseAsJson(version=version) self.site = mwclient.Site('test.wikipedia.org') responses.reset() def tearDown(self): mock.patch.stopall() def test_token_new_system(self): # Test get_token for MW >= 1.24 self.configure(version='1.24') self.raw_call.return_value = json.dumps({ 'query': {'tokens': {'csrftoken': 'sometoken'}} }) self.site.get_token('edit') args, kwargs = self.raw_call.call_args data = args[1] assert 'intoken' not in data assert data.get('type') == 'csrf' assert 'csrf' in self.site.tokens assert self.site.tokens['csrf'] == 'sometoken' assert 'edit' not in self.site.tokens def test_token_old_system_without_specifying_title(self): # Test get_token for MW < 1.24 self.configure(version='1.23') self.raw_call.return_value = self.makePageResponse(edittoken='sometoken', title='Test') self.site.get_token('edit') args, kwargs = self.raw_call.call_args data = args[1] assert 'type' not in data assert data.get('intoken') == 'edit' assert 'edit' in self.site.tokens assert self.site.tokens['edit'] == 'sometoken' assert 'csrf' not in self.site.tokens def test_token_old_system_with_specifying_title(self): # Test get_token for MW < 1.24 self.configure(version='1.23') self.raw_call.return_value = self.makePageResponse(edittoken='sometoken', title='Some page') self.site.get_token('edit', title='Some page') args, kwargs = self.raw_call.call_args data = args[1] assert self.site.tokens['edit'] == 'sometoken' class TestClientPatrol(TestCase): def setUp(self): self.raw_call = mock.patch('mwclient.client.Site.raw_call').start() def configure(self, version='1.24'): self.raw_call.return_value = self.metaResponseAsJson(version=version) self.site = mwclient.Site('test.wikipedia.org') def tearDown(self): mock.patch.stopall() @mock.patch('mwclient.client.Site.get_token') def test_patrol(self, get_token): self.configure('1.24') get_token.return_value = 'sometoken' patrol_response = {"patrol": {"rcid": 12345, "ns": 0, "title": "Foo"}} self.raw_call.return_value = json.dumps(patrol_response) resp = self.site.patrol(12345) assert resp == patrol_response["patrol"] get_token.assert_called_once_with('patrol') @mock.patch('mwclient.client.Site.get_token') def test_patrol_on_mediawiki_below_1_17(self, get_token): self.configure('1.16') get_token.return_value = 'sometoken' patrol_response = {"patrol": {"rcid": 12345, "ns": 0, "title": "Foo"}} self.raw_call.return_value = json.dumps(patrol_response) resp = self.site.patrol(12345) assert resp == patrol_response["patrol"] get_token.assert_called_once_with('edit') class TestUser(TestCase): @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_create_user(self, raw_api, site_init): createaccount_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'createaccounttoken': createaccount_token}} } elif 'username' in kwargs: assert kwargs['createtoken'] == createaccount_token assert kwargs['retype'] == kwargs['password'] assert kwargs.get('createreturnurl') return { 'createaccount': {'status': 'PASS'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') password = 'password' url = '%s://%s' % (site.scheme, site.host) site.create_user(username='myusername', password=password) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='createaccount') assert call_args[1] == mock.call('createaccount', 'POST', username='myusername', password=password, retype=password, createreturnurl=url, createtoken=createaccount_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_create_user_fail_badretype(self, raw_api, site_init): createaccount_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'createaccounttoken': createaccount_token}} } elif 'username' in kwargs: assert kwargs['createtoken'] == createaccount_token assert kwargs['retype'] != kwargs['password'] assert kwargs.get('createreturnurl') return { 'createaccount': {'status': 'FAIL', 'messagecode': 'badretype', 'message': 'oups'} } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') password = 'password' url = '%s://%s' % (site.scheme, site.host) with pytest.raises(mwclient.errors.UserCreateError): site.create_user(username='myusername', password=password, retype=password[::-1]) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='createaccount') assert call_args[1] == mock.call('createaccount', 'POST', username='myusername', password=password, retype=password[::-1], createreturnurl=url, createtoken=createaccount_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user(self, raw_api, site_init): def side_effect(*args, **kwargs): if kwargs.get('list') == 'users': return { 'query': { 'users': [{ 'userid': 1, 'user': 'myusername', 'groups': ['*', 'user'] }] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.get_user(username='myusername') call_kwargs = { 'ususers': 'myusername', 'list': 'users', 'continue': '', 'meta': 'userinfo', 'uiprop': 'blockinfo|hasmsg', 'usprop': 'registration|groups|blockinfo' } call_args = raw_api.call_args_list assert len(call_args) == 1 assert call_args[0] == mock.call('query', 'GET', **call_kwargs) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user_fail_notfound(self, raw_api, site_init): def side_effect(*args, **kwargs): if kwargs.get('list') == 'users': ret = { 'query': { 'users': [{ 'missing': '' }] } } if 'ususers' in kwargs: ret['query']['users'][0]['user'] = kwargs['ususers'] elif 'ususerids' in kwargs: ret['query']['users'][0]['userid'] = kwargs['ususerids'] return ret raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') with pytest.raises(mwclient.errors.UserNotFound): site.get_user(username='notfounduser') with pytest.raises(mwclient.errors.UserNotFound): site.get_user(userid=42) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user_fail_params(self, raw_api, site_init): site = mwclient.Site('test.wikipedia.org') with pytest.raises(ValueError): site.get_user(username=None, userid=None) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_block_user(self, raw_api, site_init): csrf_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'csrftoken': csrf_token}} } else: return { 'block': { 'user': 'myusername', 'userID': 1, 'expiry': 'infinite', 'id': 1, 'reason': kwargs['reason'] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.block_user(username='myusername', reason='Test', tags=['knock', 'knock']) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='csrf') assert call_args[1] == mock.call('block', 'POST', user='myusername', reason='Test', tags='knock|knock', token=csrf_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_unblock_user(self, raw_api, site_init): csrf_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'csrftoken': csrf_token}} } else: return { 'id': 1, 'user': 'myusername', 'userID': 1, 'reason': kwargs['reason'] } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.unblock_user(username='myusername', reason='Test') call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='csrf') assert call_args[1] == mock.call('unblock', 'POST', user='myusername', reason='Test', token=csrf_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_block_user_fail_params(self, raw_api, site_init): site = mwclient.Site('test.wikipedia.org') with pytest.raises(ValueError): site.block_user(reason='Test', tags=['knock', 'knock']) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_block_user_fail_unkown(self, raw_api, site_init): csrf_token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'csrftoken': csrf_token}} } else: return { } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') with pytest.raises(mwclient.errors.APIError): site.block_user(userid=42, reason='Test', tags=['knock', 'knock']) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='csrf') assert call_args[1] == mock.call('block', 'POST', userid=42, reason='Test', tags='knock|knock', token=csrf_token) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user_groups(self, raw_api, site_init): def side_effect(*args, **kwargs): if kwargs.get('list') == 'users': return { 'query': { 'users': [{ 'userid': 1, 'user': 'myusername', 'groups': ['*', 'user'] }] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') groups = site.get_user_groups(username='myusername') call_kwargs = { 'ususers': 'myusername', 'list': 'users', 'continue': '', 'meta': 'userinfo', 'uiprop': 'blockinfo|hasmsg', 'usprop': 'groups' } call_args = raw_api.call_args_list assert len(call_args) == 1 assert call_args[0] == mock.call('query', 'GET', **call_kwargs) assert groups == ['*', 'user'] @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user_groups_fail_params(self, raw_api, site_init): site = mwclient.Site('test.wikipedia.org') with pytest.raises(ValueError): site.get_user_groups(username=None) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_get_user_groups_fail_notfound(self, raw_api, site_init): def side_effect(*args, **kwargs): if kwargs.get('list') == 'users': return { 'query': { 'users': [{ 'userid': 42, 'missing': '' }] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') with pytest.raises(mwclient.errors.UserNotFound): site.get_user_groups(userid=42) call_kwargs = { 'ususerids': 42, 'list': 'users', 'continue': '', 'meta': 'userinfo', 'uiprop': 'blockinfo|hasmsg', 'usprop': 'groups' } call_args = raw_api.call_args_list assert len(call_args) == 1 assert call_args[0] == mock.call('query', 'GET', **call_kwargs) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_add_user_groups(self, raw_api, site_init): token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'userrightstoken': token}} } else: return { 'userrights': { 'userid': 42, 'user': 'myusername', 'added': ['*', 'user', 'bureaucrat', 'sysop'], 'removed': [] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') added = site.add_user_groups(username='myusername', groups=['*', 'user', 'bureaucrat', 'sysop']) mock_call = mock.call('userrights', 'POST', **{ 'user': 'myusername', 'add': '*|user|bureaucrat|sysop', 'token': token, }) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='userrights') mock_call_kwargs = deepcopy(mock_call[2]) real_call_kwargs = deepcopy(call_args[1][1]) assert 'add' in real_call_kwargs assert 'add' in mock_call_kwargs add_kwargs = set(real_call_kwargs.pop('add').split('|')) assert add_kwargs == set(mock_call_kwargs.pop('add').split('|')) assert 'remove' not in real_call_kwargs assert 'remove' not in mock_call_kwargs if sys.version_info >= (3,8,0): assert real_call_kwargs == mock_call_kwargs assert mock_call.args == call_args[1].args assert added == ['*', 'user', 'bureaucrat', 'sysop'] @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_add_user_groups_expirty_formats(self, raw_api, site_init): token = 'abc+\\' today = date.today() def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'userrightstoken': token}} } else: return { 'userrights': { 'userid': 42, 'user': 'myusername', 'added': ['*', 'user', 'bureaucrat', 'sysop'], 'removed': [] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') call_kwargs = { 'username': 'myusername', 'groups': ['*', 'user', 'bureaucrat', 'sysop'], } mock_call_kwargs = { 'user': 'myusername', 'add': '*|user|bureaucrat|sysop', 'token': token, } expirty_formats = ( (['2042-01-01', '2042-01-02'], '2042-01-01|2042-01-02'), ('2042-01-01', '2042-01-01'), (today, '%s' % today), ) for call_fmt_expiry, mock_call_expiry in expirty_formats: call_kwargs['expiry'] = call_fmt_expiry mock_call_kwargs['expiry'] = mock_call_expiry added = site.add_user_groups(**call_kwargs) assert added == ['*', 'user', 'bureaucrat', 'sysop'] real_mock_call_kwargs = mock.call('userrights', 'POST', **mock_call_kwargs)[2] real_call_kwargs = raw_api.call_args_list[-1][1] assert real_mock_call_kwargs['expiry'] == real_call_kwargs['expiry'] @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_add_user_groups_fail_notfound(self, raw_api, site_init): token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'userrightstoken': token}} } else: raise mwclient.errors.APIError('nosuchuser', 'Blah', kwargs) raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') with pytest.raises(mwclient.errors.UserNotFound): site.add_user_groups(username='notfound', groups=['*', 'user', 'bureaucrat', 'sysop']) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_add_user_groups_fail_params(self, raw_api, site_init): site = mwclient.Site('test.wikipedia.org') with pytest.raises(ValueError): site.add_user_groups(username=None, groups=['*', 'user', 'bureaucrat', 'sysop']) assert [] == site.add_user_groups(userid=42, groups=[]) @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_remove_user_groups(self, raw_api, site_init): token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'userrightstoken': token}} } else: return { 'userrights': { 'userid': 42, 'user': 'myusername', 'removed': ['bureaucrat', 'sysop'], 'added': [] } } raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') removed = site.remove_user_groups(username='myusername', groups=['bureaucrat', 'sysop'], reason='Test') mock_call = mock.call('userrights', 'POST', **{ 'user': 'myusername', 'remove': 'bureaucrat|sysop', 'token': token, 'reason': 'Test', }) call_args = raw_api.call_args_list assert len(call_args) == 2 assert call_args[0] == mock.call('query', 'GET', meta='tokens', type='userrights') mock_call_kwargs = deepcopy(mock_call[2]) real_call_kwargs = deepcopy(call_args[1][1]) assert 'remove' in real_call_kwargs assert 'remove' in mock_call_kwargs remove_kwargs = set(real_call_kwargs.pop('remove').split('|')) assert remove_kwargs == set(mock_call_kwargs.pop('remove').split('|')) assert 'add' not in real_call_kwargs assert 'add' not in mock_call_kwargs if sys.version_info >= (3,8,0): assert real_call_kwargs == mock_call_kwargs assert mock_call.args == call_args[1].args assert removed == ['bureaucrat', 'sysop'] @mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.raw_api') def test_set_user_groups(self, raw_api, site_init): token = 'abc+\\' def side_effect(*args, **kwargs): if kwargs.get('meta') == 'tokens': return { 'query': {'tokens': {'userrightstoken': token}} } elif kwargs.get('list') == 'users': return { 'query': { 'users': [{ 'userid': 42, 'user': 'myusername', 'groups': ['*', 'user', 'bot', 'interface-admin'] }] } } else: return {} raw_api.side_effect = side_effect site = mwclient.Site('test.wikipedia.org') site.set_user_groups(userid=42, tags=['one', 'two'], reason='Test', groups=['*', 'user', 'bureaucrat', 'sysop']) get_groups_call_kwargs = { 'ususerids': 42, 'list': 'users', 'continue': '', 'meta': 'userinfo', 'uiprop': 'blockinfo|hasmsg', 'usprop': 'groups' } set_groups_call_kwargs = { 'userid': 42, 'remove': 'bot|interface-admin', 'add': 'bureaucrat|sysop', 'reason': 'Test', 'tags': 'one|two', 'token': token, } call_args = raw_api.call_args_list assert len(call_args) == 3 assert call_args[0] == mock.call('query', 'GET', **get_groups_call_kwargs) assert call_args[1] == mock.call('query', 'GET', meta='tokens', type='userrights') mock_call = mock.call('userrights', 'POST', **set_groups_call_kwargs) mock_call_kwargs = deepcopy(mock_call[2]) real_call_kwargs = deepcopy(call_args[2][1]) assert 'add' in real_call_kwargs assert 'add' in mock_call_kwargs add_kwargs = set(real_call_kwargs.pop('add').split('|')) assert add_kwargs == set(mock_call_kwargs.pop('add').split('|')) assert 'remove' in real_call_kwargs assert 'remove' in mock_call_kwargs remove_kwargs = set(real_call_kwargs.pop('remove').split('|')) assert remove_kwargs == set(mock_call_kwargs.pop('remove').split('|')) if sys.version_info >= (3,8,0): assert real_call_kwargs == mock_call_kwargs assert mock_call.args == call_args[2].args if __name__ == '__main__': unittest.main()