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 index 0000000..f1a15ea --- /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 index 946bd78..00e44e8 100644 --- a/swh/core/pytest_plugin.py +++ b/swh/core/pytest_plugin.py @@ -1,207 +1,294 @@ # Copyright (C) 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 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__) # Check get_local_factory function # Maximum number of iteration checks to generate requests responses MAX_VISIT_FILES = 10 def get_response_cb(request, context, datadir, ignore_urls: List[str] = [], visits: Optional[Dict] = None): """Mount point callback to fetch on disk the request's content. This is meant to be used as 'body' argument of the requests_mock.get() method. It will look for files on the local filesystem based on the requested URL, using the following rules: - files are searched in the datadir/ directory - the local file name is the path part of the URL with path hierarchy markers (aka '/') replaced by '_' Eg. if you use the requests_mock fixture in your test file as: requests_mock.get('https://nowhere.com', body=get_response_cb) # or even requests_mock.get(re.compile('https://'), body=get_response_cb) then a call requests.get like: requests.get('https://nowhere.com/path/to/resource?a=b&c=d') will look the content of the response in: datadir/nowhere.com/path_to_resource,a=b,c=d Args: request (requests.Request): Object requests context (requests.Context): Object holding response metadata information (status_code, headers, etc...) ignore_urls: urls whose status response should be 404 even if the local file exists visits: Dict of url, number of visits. If None, disable multi visit support (default) Returns: Optional[FileDescriptor] on disk file to read from the test context """ logger.debug('get_response_cb(%s, %s)', request, context) logger.debug('url: %s', request.url) logger.debug('ignore_urls: %s', ignore_urls) if request.url in ignore_urls: context.status_code = 404 return None url = urlparse(request.url) dirname = url.hostname # pypi.org | files.pythonhosted.org # url.path: pypi//json -> local file: pypi__json filename = url.path[1:] if filename.endswith('/'): filename = filename[:-1] filename = filename.replace('/', '_') if url.query: filename += ',' + url.query.replace('&', ',') filepath = path.join(datadir, dirname, filename) if visits is not None: visit = visits.get(url, 0) visits[url] = visit + 1 if visit: filepath = filepath + '_visit%s' % visit if not path.isfile(filepath): logger.debug('not found filepath: %s', filepath) context.status_code = 404 return None fd = open(filepath, 'rb') context.headers['content-length'] = str(path.getsize(filepath)) return fd @pytest.fixture def datadir(request): """By default, returns the test directory's data directory. This can be overridden on a per arborescence basis. Add an override definition in the local conftest, for example: import pytest from os import path @pytest.fixture def datadir(): return path.join(path.abspath(path.dirname(__file__)), 'resources') """ return path.join(path.dirname(str(request.fspath)), 'data') def requests_mock_datadir_factory(ignore_urls: List[str] = [], has_multi_visit: bool = False): """This factory generates fixture which allow to look for files on the local filesystem based on the requested URL, using the following rules: - files are searched in the datadir/ directory - the local file name is the path part of the URL with path hierarchy markers (aka '/') replaced by '_' Multiple implementations are possible, for example: - requests_mock_datadir_factory([]): This computes the file name from the query and always returns the same result. - requests_mock_datadir_factory(has_multi_visit=True): This computes the file name from the query and returns the content of the filename the first time, the next call returning the content of files suffixed with _visit1 and so on and so forth. If the file is not found, returns a 404. - requests_mock_datadir_factory(ignore_urls=['url1', 'url2']): This will ignore any files corresponding to url1 and url2, always returning 404. Args: ignore_urls: List of urls to always returns 404 (whether file exists or not) has_multi_visit: Activate or not the multiple visits behavior """ @pytest.fixture def requests_mock_datadir(requests_mock, datadir): if not has_multi_visit: cb = partial(get_response_cb, ignore_urls=ignore_urls, datadir=datadir) requests_mock.get(re.compile('https://'), body=cb) else: visits = {} requests_mock.get(re.compile('https://'), body=partial( get_response_cb, ignore_urls=ignore_urls, visits=visits, datadir=datadir) ) return requests_mock return requests_mock_datadir # Default `requests_mock_datadir` implementation requests_mock_datadir = requests_mock_datadir_factory([]) # Implementation for multiple visits behavior: # - first time, it checks for a file named `filename` # - second time, it checks for a file named `filename`_visit1 # etc... requests_mock_datadir_visits = requests_mock_datadir_factory( 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: yield client # stolen from pytest-flask, required to have url_for() working within tests # using flask_app_client fixture. @pytest.fixture(autouse=True) def _push_request_context(request): """During tests execution request context has been pushed, e.g. `url_for`, `session`, etc. can be used in tests as is:: def test_app(app, client): assert client.get(url_for('myview')).status_code == 200 """ if 'app' not in request.fixturenames: return app = request.getfixturevalue('app') ctx = app.test_request_context() ctx.push() def teardown(): ctx.pop() request.addfinalizer(teardown)