Changeset View
Changeset View
Standalone View
Standalone View
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] |