Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_client.py
# Copyright (C) 2018-2022 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 | ||||
from swh.core.api import APIError, RemoteException, RPCClient, remote_api_endpoint | from swh.core.api import ( | ||||
APIError, | |||||
RemoteException, | |||||
RPCClient, | |||||
TransientRemoteException, | |||||
remote_api_endpoint, | |||||
) | |||||
from swh.core.api.serializers import exception_to_dict, msgpack_dumps | from swh.core.api.serializers import exception_to_dict, msgpack_dumps | ||||
from .test_serializers import ExtraType, extra_decoders, extra_encoders | from .test_serializers import ExtraType, extra_decoders, extra_encoders | ||||
class ReraiseException(Exception): | class ReraiseException(Exception): | ||||
pass | pass | ||||
▲ Show 20 Lines • Show All 118 Lines • ▼ Show 20 Lines | def test_client_reraise_exception(rpc_client, requests_mock): | ||||
) | ) | ||||
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("status_code", [400, 500]) | @pytest.mark.parametrize("status_code", [400, 500, 503]) | ||||
def test_client_raise_remote_exception(rpc_client, requests_mock, status_code): | def test_client_raise_remote_exception(rpc_client, requests_mock, status_code): | ||||
""" | """ | ||||
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, | ||||
), | ), | ||||
) | ) | ||||
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 | ||||
if status_code == 503: | |||||
assert isinstance(exc_info.value, TransientRemoteException) | |||||
else: | |||||
assert not isinstance(exc_info.value, TransientRemoteException) |