diff --git a/swh/core/api/tests/test_api.py b/swh/core/api/tests/test_api.py deleted file mode 100644 index a84030d..0000000 --- a/swh/core/api/tests/test_api.py +++ /dev/null @@ -1,86 +0,0 @@ -# Copyright (C) 2018 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 unittest - -import requests_mock -from werkzeug.wrappers import BaseResponse -from werkzeug.test import Client as WerkzeugTestClient - -from swh.core.api import ( - error_handler, encode_data_server, - remote_api_endpoint, RPCClient, RPCServerApp) - - -class ApiTest(unittest.TestCase): - def test_server(self): - testcase = self - nb_endpoint_calls = 0 - - class TestStorage: - @remote_api_endpoint('test_endpoint_url') - def test_endpoint(self, test_data, db=None, cur=None): - nonlocal nb_endpoint_calls - nb_endpoint_calls += 1 - - testcase.assertEqual(test_data, 'spam') - return 'egg' - - app = RPCServerApp('testapp', - backend_class=TestStorage, - backend_factory=lambda: TestStorage()) - - @app.errorhandler(Exception) - def my_error_handler(exception): - return error_handler(exception, encode_data_server) - - client = WerkzeugTestClient(app, BaseResponse) - res = client.post( - '/test_endpoint_url', - headers=[('Content-Type', 'application/x-msgpack'), - ('Accept', 'application/x-msgpack')], - data=b'\x81\xa9test_data\xa4spam') - - self.assertEqual(nb_endpoint_calls, 1) - self.assertEqual(b''.join(res.response), b'\xa3egg') - - def test_client(self): - class TestStorage: - @remote_api_endpoint('test_endpoint_url') - def test_endpoint(self, test_data, db=None, cur=None): - pass - - nb_http_calls = 0 - - def callback(request, context): - nonlocal nb_http_calls - nb_http_calls += 1 - self.assertEqual(request.headers['Content-Type'], - 'application/x-msgpack') - self.assertEqual(request.body, b'\x81\xa9test_data\xa4spam') - context.headers['Content-Type'] = 'application/x-msgpack' - context.content = b'\xa3egg' - return b'\xa3egg' - - adapter = requests_mock.Adapter() - adapter.register_uri('POST', - 'mock://example.com/test_endpoint_url', - content=callback) - - class Testclient(RPCClient): - backend_class = TestStorage - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # we need to mount the mock adapter on the base url to override - # RPCClient's mechanism that also mounts an HTTPAdapter - # (for configuration purpose) - self.session.mount('mock://example.com/', adapter) - - c = Testclient(url='mock://example.com/') - res = c.test_endpoint('spam') - - self.assertEqual(nb_http_calls, 1) - self.assertEqual(res, 'egg') diff --git a/swh/core/api/tests/test_rpc_client.py b/swh/core/api/tests/test_rpc_client.py new file mode 100644 index 0000000..307a5e7 --- /dev/null +++ b/swh/core/api/tests/test_rpc_client.py @@ -0,0 +1,56 @@ +# 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 re +import pytest + +from swh.core.api import remote_api_endpoint, RPCClient + + +@pytest.fixture +def rpc_client(requests_mock): + class TestStorage: + @remote_api_endpoint('test_endpoint_url') + def test_endpoint(self, test_data, db=None, cur=None): + return 'egg' + + @remote_api_endpoint('path/to/endpoint') + def something(self, data, db=None, cur=None): + return 'spam' + + class Testclient(RPCClient): + backend_class = TestStorage + + def callback(request, context): + assert request.headers['Content-Type'] == 'application/x-msgpack' + context.headers['Content-Type'] = 'application/x-msgpack' + if request.path == '/test_endpoint_url': + context.content = b'\xa3egg' + elif request.path == '/path/to/endpoint': + context.content = b'\xa4spam' + else: + assert False + return context.content + + requests_mock.post(re.compile('mock://example.com/'), + content=callback) + + return Testclient(url='mock://example.com') + + +def test_client(rpc_client): + + assert hasattr(rpc_client, 'test_endpoint') + assert hasattr(rpc_client, 'something') + + res = rpc_client.test_endpoint('spam') + assert res == 'egg' + res = rpc_client.test_endpoint(test_data='spam') + assert res == 'egg' + + res = rpc_client.something('whatever') + assert res == 'spam' + res = rpc_client.something(data='whatever') + assert res == 'spam' diff --git a/swh/core/api/tests/test_rpc_server.py b/swh/core/api/tests/test_rpc_server.py new file mode 100644 index 0000000..9399f62 --- /dev/null +++ b/swh/core/api/tests/test_rpc_server.py @@ -0,0 +1,73 @@ +# 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 +import json +import msgpack + +from flask import url_for + +from swh.core.api import remote_api_endpoint, RPCServerApp + + +@pytest.fixture +def app(): + class TestStorage: + @remote_api_endpoint('test_endpoint_url') + def test_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 + + return RPCServerApp('testapp', backend_class=TestStorage) + + +def test_api_endpoint(flask_app_client): + res = flask_app_client.post( + url_for('something'), + headers=[('Content-Type', 'application/json'), + ('Accept', 'application/json')], + data=json.dumps({'data': 'toto'}), + ) + assert res.status_code == 200 + assert res.mimetype == 'application/json' + + +def test_api_nego_default(flask_app_client): + res = flask_app_client.post( + url_for('something'), + headers=[('Content-Type', 'application/json')], + data=json.dumps({'data': 'toto'}), + ) + assert res.status_code == 200 + assert res.mimetype == 'application/json' + assert res.data == b'"toto"' + + +def test_api_nego_accept(flask_app_client): + res = flask_app_client.post( + url_for('something'), + headers=[('Accept', 'application/x-msgpack'), + ('Content-Type', 'application/x-msgpack')], + data=msgpack.dumps({'data': 'toto'}), + ) + assert res.status_code == 200 + assert res.mimetype == 'application/x-msgpack' + assert res.data == b'\xa4toto' + + +def test_rpc_server(flask_app_client): + res = flask_app_client.post( + url_for('test_endpoint'), + headers=[('Content-Type', 'application/x-msgpack'), + ('Accept', 'application/x-msgpack')], + data=b'\x81\xa9test_data\xa4spam') + + assert res.status_code == 200 + assert res.mimetype == 'application/x-msgpack' + assert res.data == b'\xa3egg' diff --git a/swh/core/pytest_plugin.py b/swh/core/pytest_plugin.py index 9521c59..946bd78 100644 --- a/swh/core/pytest_plugin.py +++ b/swh/core/pytest_plugin.py @@ -1,178 +1,207 @@ # 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 from functools import partial from os import path from typing import Dict, List, Optional from urllib.parse import urlparse 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 overriden on a per arborescence basis. Add an override + 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.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)