diff --git a/requirements-test.txt b/requirements-test.txt --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1,6 +1,7 @@ Click pytest pytest-postgresql +pytest-flask requests-mock hypothesis >= 3.11.0 pre-commit 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(client): + res = 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(client): + res = 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(client): + res = 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(client): + res = 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'