diff --git a/swh/core/api/tests/test_api.py b/swh/core/api/tests/test_api.py deleted file mode 100644 --- 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 --- /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 --- /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 --- a/swh/core/pytest_plugin.py +++ b/swh/core/pytest_plugin.py @@ -100,7 +100,7 @@ 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 @@ -176,3 +176,32 @@ # 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)