Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_client_server.py
| # Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import pytest | import pytest | ||||
| from swh.core.api import remote_api_endpoint, RPCServerApp, RPCClient | from swh.core.api import remote_api_endpoint, RPCServerApp, RPCClient | ||||
| from swh.core.api import error_handler, encode_data_server | from swh.core.api import error_handler, encode_data_server, RemoteException | ||||
| # this class is used on the server part | # this class is used on the server part | ||||
| class RPCTest: | class RPCTest: | ||||
| @remote_api_endpoint('endpoint_url') | @remote_api_endpoint('endpoint_url') | ||||
| def endpoint(self, test_data, db=None, cur=None): | def endpoint(self, test_data, db=None, cur=None): | ||||
| assert test_data == 'spam' | assert test_data == 'spam' | ||||
| return 'egg' | return 'egg' | ||||
| ▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Lines | |||||
| def test_api_client_endpoint_missing(swh_rpc_client): | def test_api_client_endpoint_missing(swh_rpc_client): | ||||
| with pytest.raises(AttributeError): | with pytest.raises(AttributeError): | ||||
| swh_rpc_client.missing(data='whatever') | swh_rpc_client.missing(data='whatever') | ||||
| def test_api_server_endpoint_missing(swh_rpc_client): | def test_api_server_endpoint_missing(swh_rpc_client): | ||||
| # A 'missing' endpoint (server-side) should raise an exception | # A 'missing' endpoint (server-side) should raise an exception | ||||
| # due to a 404, since at the end, we do a GET/POST an inexistent URL | # due to a 404, since at the end, we do a GET/POST an inexistent URL | ||||
| with pytest.raises(Exception, match='404 Not Found'): | with pytest.raises(Exception, match='404 not found'): | ||||
| swh_rpc_client.not_on_server() | swh_rpc_client.not_on_server() | ||||
| def test_api_endpoint_kwargs(swh_rpc_client): | def test_api_endpoint_kwargs(swh_rpc_client): | ||||
| res = swh_rpc_client.something(data='whatever') | res = swh_rpc_client.something(data='whatever') | ||||
| assert res == 'whatever' | assert res == 'whatever' | ||||
| res = swh_rpc_client.endpoint(test_data='spam') | res = swh_rpc_client.endpoint(test_data='spam') | ||||
| assert res == 'egg' | assert res == 'egg' | ||||
| def test_api_endpoint_args(swh_rpc_client): | def test_api_endpoint_args(swh_rpc_client): | ||||
| res = swh_rpc_client.something('whatever') | res = swh_rpc_client.something('whatever') | ||||
| assert res == 'whatever' | assert res == 'whatever' | ||||
| res = swh_rpc_client.endpoint('spam') | res = swh_rpc_client.endpoint('spam') | ||||
| assert res == 'egg' | assert res == 'egg' | ||||
| def test_api_typeerror(swh_rpc_client): | def test_api_typeerror(swh_rpc_client): | ||||
| with pytest.raises(TypeError, match='Did I pass through?'): | with pytest.raises(RemoteException) as exc_info: | ||||
| swh_rpc_client.raise_typeerror() | swh_rpc_client.raise_typeerror() | ||||
| assert exc_info.value.args[0]['type'] == 'TypeError' | |||||
| assert exc_info.value.args[0]['args'] == ['Did I pass through?'] | |||||
| assert str(exc_info.value) \ | |||||
| == "<RemoteException 500 TypeError: ['Did I pass through?']>" | |||||