Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_serializers.py
| Show All 15 Lines | from swh.core.api.serializers import ( | ||||
| SWHJSONDecoder, | SWHJSONDecoder, | ||||
| SWHJSONEncoder, | SWHJSONEncoder, | ||||
| msgpack_dumps, | msgpack_dumps, | ||||
| msgpack_loads, | msgpack_loads, | ||||
| decode_response | decode_response | ||||
| ) | ) | ||||
| class ExtraType: | |||||
| def __init__(self, arg1, arg2): | |||||
| self.arg1 = arg1 | |||||
| self.arg2 = arg2 | |||||
| def __repr__(self): | |||||
| return f'ExtraType({self.arg1}, {self.arg2})' | |||||
| def __eq__(self, other): | |||||
| return (self.arg1, self.arg2) == (other.arg1, other.arg2) | |||||
| extra_encoders = [ | |||||
| (ExtraType, 'extratype', lambda o: (o.arg1, o.arg2)) | |||||
| ] | |||||
| extra_decoders = { | |||||
| '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, 231455), | ||||
| 'datetime_tz': datetime.datetime(2015, 3, 4, 18, 25, 13, 1234, | 'datetime_tz': datetime.datetime(2015, 3, 4, 18, 25, 13, 1234, | ||||
| Show All 30 Lines | def setUp(self): | ||||
| 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): | |||||
| original_data = [ExtraType('baz', self.data), 'qux'] | |||||
| data = json.dumps(original_data, cls=SWHJSONEncoder, | |||||
| extra_encoders=extra_encoders) | |||||
| self.assertEqual( | |||||
| original_data, | |||||
| json.loads( | |||||
| 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): | ||||
| data = msgpack_dumps(self.data) | data = msgpack_dumps(self.data) | ||||
| self.assertEqual(self.data, msgpack_loads(data)) | self.assertEqual(self.data, msgpack_loads(data)) | ||||
| def test_round_trip_msgpack_extra_types(self): | |||||
| original_data = [ExtraType('baz', self.data), 'qux'] | |||||
| data = msgpack_dumps(original_data, extra_encoders=extra_encoders) | |||||
| self.assertEqual( | |||||
| 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 | ||||