Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_async.py
| # Copyright (C) 2019-2020 The Software Heritage developers | # Copyright (C) 2019-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 | ||||
| import msgpack | import msgpack | ||||
| import pytest | import pytest | ||||
| from swh.core.api.asynchronous import ( | from swh.core.api.asynchronous import ( | ||||
| Response, | Response, | ||||
| RPCServerApp, | RPCServerApp, | ||||
| decode_data, | |||||
| decode_request, | decode_request, | ||||
| encode_msgpack, | encode_msgpack, | ||||
| ) | ) | ||||
| from swh.core.api.serializers import SWHJSONEncoder, msgpack_dumps | from swh.core.api.serializers import SWHJSONEncoder, json_dumps, msgpack_dumps | ||||
| pytest_plugins = ["aiohttp.pytest_plugin", "pytester"] | pytest_plugins = ["aiohttp.pytest_plugin", "pytester"] | ||||
| class TestServerException(Exception): | class TestServerException(Exception): | ||||
| pass | pass | ||||
| ▲ Show 20 Lines • Show All 199 Lines • ▼ Show 20 Lines | for ctype in ("x-msgpack", "json"): | ||||
| "Content-Type": "application/json", | "Content-Type": "application/json", | ||||
| "Accept": "application/%s" % ctype, | "Accept": "application/%s" % ctype, | ||||
| }, | }, | ||||
| data=json.dumps(STRUCT, cls=SWHJSONEncoder), | data=json.dumps(STRUCT, cls=SWHJSONEncoder), | ||||
| ) | ) | ||||
| assert resp.status == 200 | assert resp.status == 200 | ||||
| check_mimetype(resp.headers["Content-Type"], "application/x-msgpack") | check_mimetype(resp.headers["Content-Type"], "application/x-msgpack") | ||||
| assert (await decode_request(resp)) == STRUCT | assert (await decode_request(resp)) == STRUCT | ||||
| def test_async_decode_data_failure(): | |||||
| with pytest.raises(ValueError, match="Wrong content type"): | |||||
| decode_data("some-data", "unknown-content-type") | |||||
| @pytest.mark.parametrize("data", [None, "", {}, []]) | |||||
| def test_async_decode_data_empty_cases(data): | |||||
| assert decode_data(data, "unknown-content-type") == {} | |||||
| @pytest.mark.parametrize( | |||||
| "data,content_type,encode_data_fn", | |||||
| [ | |||||
| ({"a": 1}, "application/json", json_dumps), | |||||
| ({"a": 1}, "application/x-msgpack", msgpack_dumps), | |||||
| ], | |||||
| ) | |||||
| def test_async_decode_data_nominal(data, content_type, encode_data_fn): | |||||
| actual_data = decode_data(encode_data_fn(data), content_type) | |||||
| assert actual_data == data | |||||