Page MenuHomeSoftware Heritage
Differential D2772 Diff 10088 swh/core/api/tests/test_serializers.py

Changeset View

Standalone View

View Options

swh/core/api/tests/test_serializers.py

Show All 12 Lines
import requests import requests
import requests_mock import requests_mock
from swh.core.api.serializers import ( from swh.core.api.serializers import (
SWHJSONDecoder, SWHJSONDecoder,
SWHJSONEncoder, SWHJSONEncoder,
msgpack_dumps, msgpack_dumps,
msgpack_loads, msgpack_loads,
decode_response decode_response,
) )
class ExtraType: class ExtraType:
def __init__(self, arg1, arg2): def __init__(self, arg1, arg2):
self.arg1 = arg1 self.arg1 = arg1
self.arg2 = arg2 self.arg2 = arg2
def __repr__(self): def __repr__(self):
return f'ExtraType({self.arg1}, {self.arg2})' return f"ExtraType({self.arg1}, {self.arg2})"
def __eq__(self, other): def __eq__(self, other):
return isinstance(other, ExtraType) \ return isinstance(other, ExtraType) and (self.arg1, self.arg2) == (
and (self.arg1, self.arg2) == (other.arg1, other.arg2) other.arg1,
other.arg2,
)
extra_encoders: List[Tuple[type, str, Callable[..., Any]]] = [ extra_encoders: List[Tuple[type, str, Callable[..., Any]]] = [
(ExtraType, 'extratype', lambda o: (o.arg1, o.arg2)) (ExtraType, "extratype", lambda o: (o.arg1, o.arg2))
] ]
extra_decoders = { extra_decoders = {"extratype": lambda o: ExtraType(*o)}
'extratype': lambda o: ExtraType(*o),
}
class Serializers(unittest.TestCase): class Serializers(unittest.TestCase):
def setUp(self): def setUp(self):
self.tz = datetime.timezone(datetime.timedelta(minutes=118)) self.tz = datetime.timezone(datetime.timedelta(minutes=118))
self.data = { self.data = {
'bytes': b'123456789\x99\xaf\xff\x00\x12', "bytes": b"123456789\x99\xaf\xff\x00\x12",
'datetime_naive': datetime.datetime(2015, 1, 1, 12, 4, 42, 231455), "datetime_naive": datetime.datetime(2015, 1, 1, 12, 4, 42, 231_455),
'datetime_tz': datetime.datetime(2015, 3, 4, 18, 25, 13, 1234, "datetime_tz": datetime.datetime(
tzinfo=self.tz), 2015, 3, 4, 18, 25, 13, 1234, tzinfo=self.tz
'datetime_utc': datetime.datetime(2015, 3, 4, 18, 25, 13, 1234, ),
tzinfo=datetime.timezone.utc), "datetime_utc": datetime.datetime(
'datetime_delta': datetime.timedelta(64), 2015, 3, 4, 18, 25, 13, 1234, tzinfo=datetime.timezone.utc
'arrow_date': arrow.get('2018-04-25T16:17:53.533672+00:00'), ),
'swhtype': 'fake', "datetime_delta": datetime.timedelta(64),
'swh_dict': {'swhtype': 42, 'd': 'test'}, "arrow_date": arrow.get("2018-04-25T16:17:53.533672+00:00"),
'random_dict': {'swhtype': 43}, "swhtype": "fake",
'uuid': UUID('cdd8f804-9db6-40c3-93ab-5955d3836234'), "swh_dict": {"swhtype": 42, "d": "test"},
"random_dict": {"swhtype": 43},
"uuid": UUID("cdd8f804-9db6-40c3-93ab-5955d3836234"),
} }
self.encoded_data = { self.encoded_data = {
'bytes': {'swhtype': 'bytes', 'd': 'F)}kWH8wXmIhn8j01^'}, "bytes": {"swhtype": "bytes", "d": "F)}kWH8wXmIhn8j01^"},
'datetime_naive': {'swhtype': 'datetime', "datetime_naive": {
'd': '2015-01-01T12:04:42.231455'}, "swhtype": "datetime",
'datetime_tz': {'swhtype': 'datetime', "d": "2015-01-01T12:04:42.231455",
'd': '2015-03-04T18:25:13.001234+01:58'}, },
'datetime_utc': {'swhtype': 'datetime', "datetime_tz": {
'd': '2015-03-04T18:25:13.001234+00:00'}, "swhtype": "datetime",
'datetime_delta': {'swhtype': 'timedelta', "d": "2015-03-04T18:25:13.001234+01:58",
'd': {'days': 64, 'seconds': 0, },
'microseconds': 0}}, "datetime_utc": {
'arrow_date': {'swhtype': 'arrow', "swhtype": "datetime",
'd': '2018-04-25T16:17:53.533672+00:00'}, "d": "2015-03-04T18:25:13.001234+00:00",
'swhtype': 'fake', },
'swh_dict': {'swhtype': 42, 'd': 'test'}, "datetime_delta": {
'random_dict': {'swhtype': 43}, "swhtype": "timedelta",
'uuid': {'swhtype': 'uuid', "d": {"days": 64, "seconds": 0, "microseconds": 0},
'd': 'cdd8f804-9db6-40c3-93ab-5955d3836234'}, },
"arrow_date": {"swhtype": "arrow", "d": "2018-04-25T16:17:53.533672+00:00"},
"swhtype": "fake",
"swh_dict": {"swhtype": 42, "d": "test"},
"random_dict": {"swhtype": 43},
"uuid": {"swhtype": "uuid", "d": "cdd8f804-9db6-40c3-93ab-5955d3836234"},
} }
self.legacy_msgpack = { self.legacy_msgpack = {
'bytes': b'\xc4\x0e123456789\x99\xaf\xff\x00\x12', "bytes": b"\xc4\x0e123456789\x99\xaf\xff\x00\x12",
'datetime_naive': ( "datetime_naive": (
b'\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xba' b"\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xba"
b'2015-01-01T12:04:42.231455' b"2015-01-01T12:04:42.231455"
), ),
'datetime_tz': ( "datetime_tz": (
b'\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xd9 ' b"\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xd9 "
b'2015-03-04T18:25:13.001234+01:58' b"2015-03-04T18:25:13.001234+01:58"
), ),
'datetime_utc': ( "datetime_utc": (
b'\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xd9 ' b"\x82\xc4\x0c__datetime__\xc3\xc4\x01s\xd9 "
b'2015-03-04T18:25:13.001234+00:00' b"2015-03-04T18:25:13.001234+00:00"
), ),
'datetime_delta': ( "datetime_delta": (
b'\x82\xc4\r__timedelta__\xc3\xc4\x01s\x83\xa4' b"\x82\xc4\r__timedelta__\xc3\xc4\x01s\x83\xa4"
b'days@\xa7seconds\x00\xacmicroseconds\x00' b"days@\xa7seconds\x00\xacmicroseconds\x00"
), ),
'arrow_date': ( "arrow_date": (
b'\x82\xc4\t__arrow__\xc3\xc4\x01s\xd9 ' b"\x82\xc4\t__arrow__\xc3\xc4\x01s\xd9 "
b'2018-04-25T16:17:53.533672+00:00' b"2018-04-25T16:17:53.533672+00:00"
), ),
'swhtype': b'\xa4fake', "swhtype": b"\xa4fake",
'swh_dict': b'\x82\xa7swhtype*\xa1d\xa4test', "swh_dict": b"\x82\xa7swhtype*\xa1d\xa4test",
'random_dict': b'\x81\xa7swhtype+', "random_dict": b"\x81\xa7swhtype+",
'uuid': ( "uuid": (
b'\x82\xc4\x08__uuid__\xc3\xc4\x01s\xd9$' b"\x82\xc4\x08__uuid__\xc3\xc4\x01s\xd9$"
b'cdd8f804-9db6-40c3-93ab-5955d3836234' b"cdd8f804-9db6-40c3-93ab-5955d3836234"
), ),
} }
self.generator = (i for i in range(5)) self.generator = (i for i in range(5))
self.gen_lst = list(range(5)) self.gen_lst = list(range(5))
def test_round_trip_json(self): def test_round_trip_json(self):
data = json.dumps(self.data, cls=SWHJSONEncoder) data = json.dumps(self.data, cls=SWHJSONEncoder)
self.assertEqual(self.data, json.loads(data, cls=SWHJSONDecoder)) self.assertEqual(self.data, json.loads(data, cls=SWHJSONDecoder))
def test_round_trip_json_extra_types(self): def test_round_trip_json_extra_types(self):
original_data = [ExtraType('baz', self.data), 'qux'] original_data = [ExtraType("baz", self.data), "qux"]
data = json.dumps(original_data, cls=SWHJSONEncoder, data = json.dumps(
extra_encoders=extra_encoders) original_data, cls=SWHJSONEncoder, extra_encoders=extra_encoders
)
self.assertEqual( self.assertEqual(
original_data, original_data,
json.loads( json.loads(data, cls=SWHJSONDecoder, extra_decoders=extra_decoders),
data, cls=SWHJSONDecoder, extra_decoders=extra_decoders)) )
def test_encode_swh_json(self): def test_encode_swh_json(self):
data = json.dumps(self.data, cls=SWHJSONEncoder) data = json.dumps(self.data, cls=SWHJSONEncoder)
self.assertEqual(self.encoded_data, json.loads(data)) self.assertEqual(self.encoded_data, json.loads(data))
def test_round_trip_msgpack(self): def test_round_trip_msgpack(self):
original_data = { original_data = {**self.data, "none_dict_key": {None: 42}}
**self.data,
'none_dict_key': {None: 42},
}
data = msgpack_dumps(original_data) data = msgpack_dumps(original_data)
self.assertEqual(original_data, msgpack_loads(data)) self.assertEqual(original_data, msgpack_loads(data))
def test_round_trip_msgpack_extra_types(self): def test_round_trip_msgpack_extra_types(self):
original_data = [ExtraType('baz', self.data), 'qux'] original_data = [ExtraType("baz", self.data), "qux"]
data = msgpack_dumps(original_data, extra_encoders=extra_encoders) data = msgpack_dumps(original_data, extra_encoders=extra_encoders)
self.assertEqual( self.assertEqual(
original_data, msgpack_loads(data, extra_decoders=extra_decoders)) original_data, msgpack_loads(data, extra_decoders=extra_decoders)
)
def test_generator_json(self): def test_generator_json(self):
data = json.dumps(self.generator, cls=SWHJSONEncoder) data = json.dumps(self.generator, cls=SWHJSONEncoder)
self.assertEqual(self.gen_lst, json.loads(data, cls=SWHJSONDecoder)) self.assertEqual(self.gen_lst, json.loads(data, cls=SWHJSONDecoder))
def test_generator_msgpack(self): def test_generator_msgpack(self):
data = msgpack_dumps(self.generator) data = msgpack_dumps(self.generator)
self.assertEqual(self.gen_lst, msgpack_loads(data)) self.assertEqual(self.gen_lst, msgpack_loads(data))
@requests_mock.Mocker() @requests_mock.Mocker()
def test_decode_response_json(self, mock_requests): def test_decode_response_json(self, mock_requests):
mock_requests.get('https://example.org/test/data', mock_requests.get(
"https://example.org/test/data",
json=self.encoded_data, json=self.encoded_data,
headers={'content-type': 'application/json'}) headers={"content-type": "application/json"},
response = requests.get('https://example.org/test/data') )
response = requests.get("https://example.org/test/data")
assert decode_response(response) == self.data assert decode_response(response) == self.data
def test_decode_legacy_msgpack(self): def test_decode_legacy_msgpack(self):
for k, v in self.legacy_msgpack.items(): for k, v in self.legacy_msgpack.items():
assert msgpack_loads(v) == self.data[k] assert msgpack_loads(v) == self.data[k]
About · Developer information · Legal