diff --git a/swh/core/api/tests/test_rpc_client_server.py b/swh/core/api/tests/test_rpc_client_server.py new file mode 100644 --- /dev/null +++ b/swh/core/api/tests/test_rpc_client_server.py @@ -0,0 +1,89 @@ +# Copyright (C) 2018-2019 The Software Heritage developers +# See the AUTHORS file at the top-level directory of this distribution +# License: GNU General Public License version 3, or any later version +# See top-level LICENSE file for more information + +import pytest + +from swh.core.api import remote_api_endpoint, RPCServerApp, RPCClient +from swh.core.api import error_handler, encode_data_server + + +# this class is used on the server part +class RPCTest: + @remote_api_endpoint('endpoint_url') + def endpoint(self, test_data, db=None, cur=None): + assert test_data == 'spam' + return 'egg' + + @remote_api_endpoint('path/to/endpoint') + def something(self, data, db=None, cur=None): + return data + + +# this class is used on the client part. We cannot inherit from RPCTest +# because the automagic metaclass based code that generates the RPCClient +# proxy class from this does not handle inheritance properly. +# We do add an endpoint on the client side that has no implementation +# server-side to test this very situation (in should generate a 404) +class RPCTest2: + @remote_api_endpoint('endpoint_url') + def endpoint(self, test_data, db=None, cur=None): + assert test_data == 'spam' + return 'egg' + + @remote_api_endpoint('path/to/endpoint') + def something(self, data, db=None, cur=None): + return data + + @remote_api_endpoint('not_on_server') + def not_on_server(self, db=None, cur=None): + return 'ok' + + +class RPCTestClient(RPCClient): + backend_class = RPCTest2 + + +@pytest.fixture +def app(): + # This fixture is used by the 'swh_rpc_adapter' fixture + # which is defined in swh/core/pytest_plugin.py + application = RPCServerApp('testapp', backend_class=RPCTest) + @application.errorhandler(Exception) + def my_error_handler(exception): + return error_handler(exception, encode_data_server) + return application + + +@pytest.fixture +def swh_rpc_client_class(): + # This fixture is used by the 'swh_rpc_client' fixture + # which is defined in swh/core/pytest_plugin.py + return RPCTestClient + + +def test_api_client_endpoint_missing(swh_rpc_client): + with pytest.raises(AttributeError): + swh_rpc_client.missing(data='whatever') + + +def test_api_server_endpoint_missing(swh_rpc_client): + # A 'missing' endpoint (server-side) should raise an exception + # due to a 404, since at the end, we do a GET/POST an inexistant URL + with pytest.raises(Exception, match='404 Not Found'): + swh_rpc_client.not_on_server() + + +def test_api_endpoint_kwargs(swh_rpc_client): + res = swh_rpc_client.something(data='whatever') + assert res == 'whatever' + res = swh_rpc_client.endpoint(test_data='spam') + assert res == 'egg' + + +def test_api_endpoint_args(swh_rpc_client): + res = swh_rpc_client.something('whatever') + assert res == 'whatever' + res = swh_rpc_client.endpoint('spam') + assert res == 'egg' diff --git a/swh/core/pytest_plugin.py b/swh/core/pytest_plugin.py --- a/swh/core/pytest_plugin.py +++ b/swh/core/pytest_plugin.py @@ -6,12 +6,17 @@ import logging import re import pytest +import requests from functools import partial from os import path from typing import Dict, List, Optional from urllib.parse import urlparse +from requests.adapters import BaseAdapter +from requests.structures import CaseInsensitiveDict +from requests.utils import get_encoding_from_headers + logger = logging.getLogger(__name__) @@ -178,6 +183,88 @@ has_multi_visit=True) +@pytest.fixture +def swh_rpc_client(swh_rpc_client_class, swh_rpc_adapter): + """This fixture generates an RPCClient instance that uses the class generated + by the rpc_client_class fixture as backend. + + Since it uses the swh_rpc_adapter, HTTP queries will be intercepted and + routed directly to the current Flask app (as provided by the `app` + fixture). + + So this stack of fixtures allows to test the RPCClient -> RPCServerApp + communication path using a real RPCClient instance and a real Flask + (RPCServerApp) app instance. + + To use this fixture: + + - ensure an `app` fixture exists and generate a Flask application, + - implement an `swh_rpc_client_class` fixtures that returns the + RPCClient-based class to use as client side for the tests, + - implement your tests using this `swh_rpc_client` fixture. + + See swh/core/api/tests/test_rpc_client_server.py for an example of usage. + """ + url = 'mock://example.com' + cli = swh_rpc_client_class(url=url) + # we need to clear the list of existing adapters here so we ensure we + # have one and only one adapter which is then used for all the requests. + cli.session.adapters.clear() + cli.session.mount('mock://', swh_rpc_adapter) + return cli + + +@pytest.yield_fixture +def swh_rpc_adapter(app): + """Fixture that generates a requests.Adapter instance that + can be used to test client/servers code based on swh.core.api classes. + + See swh/core/api/tests/test_rpc_client_server.py for an example of usage. + + """ + with app.test_client() as client: + yield RPCTestAdapter(client) + + +class RPCTestAdapter(BaseAdapter): + def __init__(self, client): + self._client = client + + def build_response(self, req, resp): + response = requests.Response() + + # Fallback to None if there's no status_code, for whatever reason. + response.status_code = resp.status_code + + # Make headers case-insensitive. + response.headers = CaseInsensitiveDict(getattr(resp, 'headers', {})) + + # Set encoding. + response.encoding = get_encoding_from_headers(response.headers) + response.raw = resp + response.reason = response.raw.status + + if isinstance(req.url, bytes): + response.url = req.url.decode('utf-8') + else: + response.url = req.url + + # Give the Response some context. + response.request = req + response.connection = self + response._content = resp.data + + return response + + def send(self, request, **kw): + resp = self._client.open( + request.url, method=request.method, + headers=request.headers.items(), + data=request.body, + ) + return self.build_response(request, resp) + + @pytest.yield_fixture def flask_app_client(app): with app.test_client() as client: