Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_client.py
| # Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2022 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 re | import re | ||||
| import pytest | import pytest | ||||
| from requests.exceptions import ConnectionError | from requests.exceptions import ConnectionError | ||||
| ▲ Show 20 Lines • Show All 95 Lines • ▼ Show 20 Lines | def test_client_connexion_error(rpc_client, requests_mock): | ||||
| with pytest.raises(APIError) as exc_info: | with pytest.raises(APIError) as exc_info: | ||||
| rpc_client._post("connection_error", data={}) | rpc_client._post("connection_error", data={}) | ||||
| assert type(exc_info.value.args[0]) == ConnectionError | assert type(exc_info.value.args[0]) == ConnectionError | ||||
| assert str(exc_info.value.args[0]) == error_message | assert str(exc_info.value.args[0]) == error_message | ||||
| def _exception_response(exception, status_code, old_exception_schema=False): | def _exception_response(exception, status_code): | ||||
| def callback(request, context): | def callback(request, context): | ||||
| assert request.headers["Content-Type"] == "application/x-msgpack" | assert request.headers["Content-Type"] == "application/x-msgpack" | ||||
| context.headers["Content-Type"] = "application/x-msgpack" | context.headers["Content-Type"] = "application/x-msgpack" | ||||
| exc_dict = exception_to_dict(exception) | exc_dict = exception_to_dict(exception) | ||||
| if old_exception_schema: | |||||
| exc_dict = {"exception": exc_dict} | |||||
| context.content = msgpack_dumps(exc_dict) | context.content = msgpack_dumps(exc_dict) | ||||
| context.status_code = status_code | context.status_code = status_code | ||||
| return context.content | return context.content | ||||
| return callback | return callback | ||||
| @pytest.mark.parametrize("old_exception_schema", [False, True]) | def test_client_reraise_exception(rpc_client, requests_mock): | ||||
| def test_client_reraise_exception(rpc_client, requests_mock, old_exception_schema): | |||||
| """ | """ | ||||
| Exception caught server-side and whitelisted will be raised again client-side. | Exception caught server-side and whitelisted will be raised again client-side. | ||||
| """ | """ | ||||
| error_message = "something went wrong" | error_message = "something went wrong" | ||||
| endpoint = "reraise_exception" | endpoint = "reraise_exception" | ||||
| requests_mock.post( | requests_mock.post( | ||||
| re.compile(f"mock://example.com/{endpoint}"), | re.compile(f"mock://example.com/{endpoint}"), | ||||
| content=_exception_response( | content=_exception_response( | ||||
| exception=ReraiseException(error_message), | exception=ReraiseException(error_message), | ||||
| status_code=400, | status_code=400, | ||||
| old_exception_schema=old_exception_schema, | |||||
| ), | ), | ||||
| ) | ) | ||||
| with pytest.raises(ReraiseException) as exc_info: | with pytest.raises(ReraiseException) as exc_info: | ||||
| rpc_client._post(endpoint, data={}) | rpc_client._post(endpoint, data={}) | ||||
| assert str(exc_info.value) == error_message | assert str(exc_info.value) == error_message | ||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize("status_code", [400, 500]) | ||||
| "status_code, old_exception_schema", | def test_client_raise_remote_exception(rpc_client, requests_mock, status_code): | ||||
| [ | |||||
| (400, False), | |||||
| (500, False), | |||||
| (400, True), | |||||
| (500, True), | |||||
| ], | |||||
| ) | |||||
| def test_client_raise_remote_exception( | |||||
| rpc_client, requests_mock, status_code, old_exception_schema | |||||
| ): | |||||
| """ | """ | ||||
| Exception caught server-side and not whitelisted will be wrapped and raised | Exception caught server-side and not whitelisted will be wrapped and raised | ||||
| as a RemoteException client-side. | as a RemoteException client-side. | ||||
| """ | """ | ||||
| error_message = "something went wrong" | error_message = "something went wrong" | ||||
| endpoint = "raise_remote_exception" | endpoint = "raise_remote_exception" | ||||
| requests_mock.post( | requests_mock.post( | ||||
| re.compile(f"mock://example.com/{endpoint}"), | re.compile(f"mock://example.com/{endpoint}"), | ||||
| content=_exception_response( | content=_exception_response( | ||||
| exception=Exception(error_message), | exception=Exception(error_message), | ||||
| status_code=status_code, | status_code=status_code, | ||||
| old_exception_schema=old_exception_schema, | |||||
| ), | ), | ||||
| ) | ) | ||||
| with pytest.raises(RemoteException) as exc_info: | with pytest.raises(RemoteException) as exc_info: | ||||
| rpc_client._post(endpoint, data={}) | rpc_client._post(endpoint, data={}) | ||||
| assert str(exc_info.value.args[0]["type"]) == "Exception" | assert str(exc_info.value.args[0]["type"]) == "Exception" | ||||
| assert str(exc_info.value.args[0]["message"]) == error_message | assert str(exc_info.value.args[0]["message"]) == error_message | ||||