Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_serializers.py
# Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-2020 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 datetime | import datetime | ||||
import json | import json | ||||
from typing import Any, Callable, List, Tuple, Union | from typing import Any, Callable, List, Tuple, Union | ||||
from uuid import UUID | from uuid import UUID | ||||
import arrow | import arrow | ||||
from arrow import Arrow | from arrow import Arrow | ||||
import pytest | import pytest | ||||
import requests | import requests | ||||
from requests.exceptions import ConnectionError | |||||
from swh.core.api.classes import PagedResult | from swh.core.api.classes import PagedResult | ||||
from swh.core.api.serializers import ( | from swh.core.api.serializers import ( | ||||
SWHJSONDecoder, | SWHJSONDecoder, | ||||
SWHJSONEncoder, | SWHJSONEncoder, | ||||
decode_response, | decode_response, | ||||
msgpack_dumps, | msgpack_dumps, | ||||
msgpack_loads, | msgpack_loads, | ||||
▲ Show 20 Lines • Show All 120 Lines • ▼ Show 20 Lines | def test_serializers_round_trip_json_extra_types(): | ||||
expected_original_data = [ExtraType("baz", DATA), "qux"] | expected_original_data = [ExtraType("baz", DATA), "qux"] | ||||
data = json.dumps( | data = json.dumps( | ||||
expected_original_data, cls=SWHJSONEncoder, extra_encoders=extra_encoders | expected_original_data, cls=SWHJSONEncoder, extra_encoders=extra_encoders | ||||
) | ) | ||||
actual_data = json.loads(data, cls=SWHJSONDecoder, extra_decoders=extra_decoders) | actual_data = json.loads(data, cls=SWHJSONDecoder, extra_decoders=extra_decoders) | ||||
assert actual_data == expected_original_data | assert actual_data == expected_original_data | ||||
def test_exception_serializer_round_trip_json(): | |||||
error_message = "unreachable host" | |||||
json_data = json.dumps( | |||||
{"exception": ConnectionError(error_message)}, cls=SWHJSONEncoder | |||||
) | |||||
actual_data = json.loads(json_data, cls=SWHJSONDecoder) | |||||
assert "exception" in actual_data | |||||
assert type(actual_data["exception"]) == ConnectionError | |||||
assert str(actual_data["exception"]) == error_message | |||||
def test_serializers_encode_swh_json(): | def test_serializers_encode_swh_json(): | ||||
json_str = json.dumps(DATA, cls=SWHJSONEncoder) | json_str = json.dumps(DATA, cls=SWHJSONEncoder) | ||||
actual_data = json.loads(json_str) | actual_data = json.loads(json_str) | ||||
assert actual_data == ENCODED_DATA | assert actual_data == ENCODED_DATA | ||||
def test_serializers_round_trip_msgpack(): | def test_serializers_round_trip_msgpack(): | ||||
expected_original_data = { | expected_original_data = { | ||||
**DATA, | **DATA, | ||||
"none_dict_key": {None: 42}, | "none_dict_key": {None: 42}, | ||||
"long_int_is_loooong": 10000000000000000000000000000000, | "long_int_is_loooong": 10000000000000000000000000000000, | ||||
} | } | ||||
data = msgpack_dumps(expected_original_data) | data = msgpack_dumps(expected_original_data) | ||||
actual_data = msgpack_loads(data) | actual_data = msgpack_loads(data) | ||||
assert actual_data == expected_original_data | assert actual_data == expected_original_data | ||||
def test_serializers_round_trip_msgpack_extra_types(): | def test_serializers_round_trip_msgpack_extra_types(): | ||||
original_data = [ExtraType("baz", DATA), "qux"] | original_data = [ExtraType("baz", DATA), "qux"] | ||||
data = msgpack_dumps(original_data, extra_encoders=extra_encoders) | data = msgpack_dumps(original_data, extra_encoders=extra_encoders) | ||||
actual_data = msgpack_loads(data, extra_decoders=extra_decoders) | actual_data = msgpack_loads(data, extra_decoders=extra_decoders) | ||||
assert actual_data == original_data | assert actual_data == original_data | ||||
def test_exception_serializer_round_trip_msgpack(): | |||||
error_message = "unreachable host" | |||||
data = msgpack_dumps({"exception": ConnectionError(error_message)}) | |||||
actual_data = msgpack_loads(data) | |||||
assert "exception" in actual_data | |||||
assert type(actual_data["exception"]) == ConnectionError | |||||
assert str(actual_data["exception"]) == error_message | |||||
def test_serializers_generator_json(): | def test_serializers_generator_json(): | ||||
data = json.dumps((i for i in range(5)), cls=SWHJSONEncoder) | data = json.dumps((i for i in range(5)), cls=SWHJSONEncoder) | ||||
assert json.loads(data, cls=SWHJSONDecoder) == [i for i in range(5)] | assert json.loads(data, cls=SWHJSONDecoder) == [i for i in range(5)] | ||||
def test_serializers_generator_msgpack(): | def test_serializers_generator_msgpack(): | ||||
data = msgpack_dumps((i for i in range(5))) | data = msgpack_dumps((i for i in range(5))) | ||||
assert msgpack_loads(data) == [i for i in range(5)] | assert msgpack_loads(data) == [i for i in range(5)] | ||||
▲ Show 20 Lines • Show All 68 Lines • Show Last 20 Lines |