test_client.py

from copy import deepcopy
from datetime import date
import json
from io import BytesIO
import logging
import sys
import time
import unittest
import unittest.mock as mock

import pytest
import requests
import responses
from requests_oauthlib import OAuth1

import mwclient

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, version='1.24wmf17', rights=None):
        if rights is None:
            rights = [
                "createaccount", "read", "edit", "createpage", "createtalk",
                "editmyusercss", "editmyuserjs", "viewmywatchlist",
                "editmywatchlist", "viewmyprivateinfo", "editmyprivateinfo",
                "editmyoptions", "centralauth-merge", "abusefilter-view",
                "abusefilter-log", "translate", "vipsscaler-test", "upload"
            ]

        # @formatter:off
        namespaces = {
            -2: {"id": -2, "*": "Media", "canonical": "Media", "case": "first-letter"},
            -1: {"id": -1, "*": "Special", "canonical": "Special", "case": "first-letter"},
            0: {"id": 0, "*": "", "case": "first-letter", "content": ""},
            1: {"id": 1, "*": "Talk", "canonical": "Talk", "case": "first-letter", "subpages": ""},
            2: {"id": 2, "*": "User", "canonical": "User", "case": "first-letter", "subpages": ""},
            3: {"id": 3, "*": "User talk", "canonical": "User talk", "case": "first-letter", "subpages": ""},
            4: {"id": 4, "*": "Wikipedia", "canonical": "Project", "case": "first-letter", "subpages": ""},
            5: {"id": 5, "*": "Wikipedia talk", "canonical": "Project talk", "case": "first-letter", "subpages": ""},
            6: {"id": 6, "*": "File", "canonical": "File", "case": "first-letter"},
            7: {"id": 7, "*": "File talk", "canonical": "File talk", "case": "first-letter", "subpages": ""},
            8: {"id": 8, "*": "MediaWiki", "canonical": "MediaWiki", "case": "first-letter", "subpages": ""},
            9: {"id": 9, "*": "MediaWiki talk", "canonical": "MediaWiki talk", "case": "first-letter", "subpages": ""},
            10: {"id": 10, "*": "Template", "canonical": "Template", "case": "first-letter", "subpages": ""},
            11: {"id": 11, "*": "Template talk", "canonical": "Template talk", "case": "first-letter", "subpages": ""},
            12: {"id": 12, "*": "Help", "canonical": "Help", "case": "first-letter", "subpages": ""},
            13: {"id": 13, "*": "Help talk", "canonical": "Help talk", "case": "first-letter", "subpages": ""},
            14: {"id": 14, "*": "Category", "canonical": "Category", "case": "first-letter"},
            15: {"id": 15, "*": "Category talk", "canonical": "Category talk", "case": "first-letter", "subpages": ""},
        }
        # @formatter:on

        return {
            "query": {
                "general": {
                    "generator": f"MediaWiki {version}"
                },
                "namespaces": namespaces,
                "userinfo": {
                    "anon": "",
                    "groups": ["*"],
                    "id": 0,
                    "name": "127.0.0.1",
                    "rights": rights
                }
            }
        }

    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 = f'{scheme}://{host}{path}{script}.php'
        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

        if sys.version_info >= (3, 8):
            import importlib.metadata

            version = importlib.metadata.version("mwclient")
        else:
            import pkg_resources  # part of setuptools

            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  # type: ignore
        assert 'retry-after' not in responses.calls[1].response.headers  # type: ignore

    @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