Differential D4515 Diff 16096 swh/core/api/tests/test_serializers.py

Changeset 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