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

Changeset View

Standalone View

View Options

swh/core/api/tests/test_async.py

Show All 9 Lines
import pytest import pytest
from swh.core.api.asynchronous import RPCServerApp, Response from swh.core.api.asynchronous import RPCServerApp, Response
from swh.core.api.asynchronous import encode_msgpack, decode_request from swh.core.api.asynchronous import encode_msgpack, decode_request
from swh.core.api.serializers import msgpack_dumps, SWHJSONEncoder from swh.core.api.serializers import msgpack_dumps, SWHJSONEncoder
pytest_plugins = ['aiohttp.pytest_plugin', 'pytester'] pytest_plugins = ["aiohttp.pytest_plugin", "pytester"]
class TestServerException(Exception): class TestServerException(Exception):
pass pass
class TestClientError(Exception): class TestClientError(Exception):
pass pass
async def root(request): async def root(request):
return Response('toor') return Response("toor")
STRUCT = {'txt': 'something stupid',
STRUCT = {
"txt": "something stupid",
# 'date': datetime.date(2019, 6, 9), # not supported # 'date': datetime.date(2019, 6, 9), # not supported
'datetime': datetime.datetime(2019, 6, 9, 10, 12), "datetime": datetime.datetime(2019, 6, 9, 10, 12),
'timedelta': datetime.timedelta(days=-2, hours=3), "timedelta": datetime.timedelta(days=-2, hours=3),
'int': 42, "int": 42,
'float': 3.14, "float": 3.14,
'subdata': {'int': 42, "subdata": {"int": 42, "datetime": datetime.datetime(2019, 6, 10, 11, 12)},
'datetime': datetime.datetime(2019, 6, 10, 11, 12), "list": [42, datetime.datetime(2019, 9, 10, 11, 12), "ok"],
},
'list': [42, datetime.datetime(2019, 9, 10, 11, 12), 'ok'],
} }
async def struct(request): async def struct(request):
return Response(STRUCT) return Response(STRUCT)
async def echo(request): async def echo(request):
data = await decode_request(request) data = await decode_request(request)
Show All 11 Lines
async def echo_no_nego(request): async def echo_no_nego(request):
# let the content negotiation handle the serialization for us... # let the content negotiation handle the serialization for us...
data = await decode_request(request) data = await decode_request(request)
ret = encode_msgpack(data) ret = encode_msgpack(data)
return ret return ret
def check_mimetype(src, dst): def check_mimetype(src, dst):
src = src.split(';')[0].strip() src = src.split(";")[0].strip()
dst = dst.split(';')[0].strip() dst = dst.split(";")[0].strip()
assert src == dst assert src == dst
@pytest.fixture @pytest.fixture
def async_app(): def async_app():
app = RPCServerApp() app = RPCServerApp()
app.client_exception_classes = (TestClientError,) app.client_exception_classes = (TestClientError,)
app.router.add_route('GET', '/', root) app.router.add_route("GET", "/", root)
app.router.add_route('GET', '/struct', struct) app.router.add_route("GET", "/struct", struct)
app.router.add_route('POST', '/echo', echo) app.router.add_route("POST", "/echo", echo)
app.router.add_route('GET', '/server_exception', server_exception) app.router.add_route("GET", "/server_exception", server_exception)
app.router.add_route('GET', '/client_error', client_error) app.router.add_route("GET", "/client_error", client_error)
app.router.add_route('POST', '/echo-no-nego', echo_no_nego) app.router.add_route("POST", "/echo-no-nego", echo_no_nego)
return app return app
async def test_get_simple(async_app, aiohttp_client) -> None: async def test_get_simple(async_app, aiohttp_client) -> None:
assert async_app is not None assert async_app is not None
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
resp = await cli.get('/') resp = await cli.get("/")
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")
data = await resp.read() data = await resp.read()
value = msgpack.unpackb(data, raw=False) value = msgpack.unpackb(data, raw=False)
assert value == 'toor' assert value == "toor"
async def test_get_server_exception(async_app, aiohttp_client) -> None: async def test_get_server_exception(async_app, aiohttp_client) -> None:
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
resp = await cli.get('/server_exception') resp = await cli.get("/server_exception")
assert resp.status == 500 assert resp.status == 500
data = await resp.read() data = await resp.read()
data = msgpack.unpackb(data, raw=False) data = msgpack.unpackb(data, raw=False)
assert data['exception']['type'] == 'TestServerException' assert data["exception"]["type"] == "TestServerException"
async def test_get_client_error(async_app, aiohttp_client) -> None: async def test_get_client_error(async_app, aiohttp_client) -> None:
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
resp = await cli.get('/client_error') resp = await cli.get("/client_error")
assert resp.status == 400 assert resp.status == 400
data = await resp.read() data = await resp.read()
data = msgpack.unpackb(data, raw=False) data = msgpack.unpackb(data, raw=False)
assert data['exception']['type'] == 'TestClientError' assert data["exception"]["type"] == "TestClientError"
async def test_get_simple_nego(async_app, aiohttp_client) -> None: async def test_get_simple_nego(async_app, aiohttp_client) -> None:
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
for ctype in ('x-msgpack', 'json'): for ctype in ("x-msgpack", "json"):
resp = await cli.get('/', headers={'Accept': 'application/%s' % ctype}) resp = await cli.get("/", headers={"Accept": "application/%s" % ctype})
assert resp.status == 200 assert resp.status == 200
check_mimetype(resp.headers['Content-Type'], 'application/%s' % ctype) check_mimetype(resp.headers["Content-Type"], "application/%s" % ctype)
assert (await decode_request(resp)) == 'toor' assert (await decode_request(resp)) == "toor"
async def test_get_struct(async_app, aiohttp_client) -> None: async def test_get_struct(async_app, aiohttp_client) -> None:
"""Test returned structured from a simple GET data is OK""" """Test returned structured from a simple GET data is OK"""
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
resp = await cli.get('/struct') resp = await cli.get("/struct")
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
async def test_get_struct_nego(async_app, aiohttp_client) -> None: async def test_get_struct_nego(async_app, aiohttp_client) -> None:
"""Test returned structured from a simple GET data is OK""" """Test returned structured from a simple GET data is OK"""
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
for ctype in ('x-msgpack', 'json'): for ctype in ("x-msgpack", "json"):
resp = await cli.get('/struct', resp = await cli.get("/struct", headers={"Accept": "application/%s" % ctype})
headers={'Accept': 'application/%s' % ctype})
assert resp.status == 200 assert resp.status == 200
check_mimetype(resp.headers['Content-Type'], 'application/%s' % ctype) check_mimetype(resp.headers["Content-Type"], "application/%s" % ctype)
assert (await decode_request(resp)) == STRUCT assert (await decode_request(resp)) == STRUCT
async def test_post_struct_msgpack(async_app, aiohttp_client) -> None: async def test_post_struct_msgpack(async_app, aiohttp_client) -> None:
"""Test that msgpack encoded posted struct data is returned as is""" """Test that msgpack encoded posted struct data is returned as is"""
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
# simple struct # simple struct
resp = await cli.post( resp = await cli.post(
'/echo', "/echo",
headers={'Content-Type': 'application/x-msgpack'}, headers={"Content-Type": "application/x-msgpack"},
data=msgpack_dumps({'toto': 42})) data=msgpack_dumps({"toto": 42}),
)
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)) == {'toto': 42} assert (await decode_request(resp)) == {"toto": 42}
# complex struct # complex struct
resp = await cli.post( resp = await cli.post(
'/echo', "/echo",
headers={'Content-Type': 'application/x-msgpack'}, headers={"Content-Type": "application/x-msgpack"},
data=msgpack_dumps(STRUCT)) data=msgpack_dumps(STRUCT),
)
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
async def test_post_struct_json(async_app, aiohttp_client) -> None: async def test_post_struct_json(async_app, aiohttp_client) -> None:
"""Test that json encoded posted struct data is returned as is""" """Test that json encoded posted struct data is returned as is"""
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
resp = await cli.post( resp = await cli.post(
'/echo', "/echo",
headers={'Content-Type': 'application/json'}, headers={"Content-Type": "application/json"},
data=json.dumps({'toto': 42}, cls=SWHJSONEncoder)) data=json.dumps({"toto": 42}, 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)) == {'toto': 42} assert (await decode_request(resp)) == {"toto": 42}
resp = await cli.post( resp = await cli.post(
'/echo', "/echo",
headers={'Content-Type': 'application/json'}, headers={"Content-Type": "application/json"},
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 resp.headers['Content-Type'] == 'application/x-msgpack' # assert resp.headers['Content-Type'] == 'application/x-msgpack'
assert (await decode_request(resp)) == STRUCT assert (await decode_request(resp)) == STRUCT
async def test_post_struct_nego(async_app, aiohttp_client) -> None: async def test_post_struct_nego(async_app, aiohttp_client) -> None:
"""Test that json encoded posted struct data is returned as is """Test that json encoded posted struct data is returned as is
using content negotiation (accept json or msgpack). using content negotiation (accept json or msgpack).
""" """
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
for ctype in ('x-msgpack', 'json'): for ctype in ("x-msgpack", "json"):
resp = await cli.post( resp = await cli.post(
'/echo', "/echo",
headers={'Content-Type': 'application/json', headers={
'Accept': 'application/%s' % ctype}, "Content-Type": "application/json",
data=json.dumps(STRUCT, cls=SWHJSONEncoder)) "Accept": "application/%s" % ctype,
},
data=json.dumps(STRUCT, cls=SWHJSONEncoder),
)
assert resp.status == 200 assert resp.status == 200
check_mimetype(resp.headers['Content-Type'], 'application/%s' % ctype) check_mimetype(resp.headers["Content-Type"], "application/%s" % ctype)
assert (await decode_request(resp)) == STRUCT assert (await decode_request(resp)) == STRUCT
async def test_post_struct_no_nego(async_app, aiohttp_client) -> None: async def test_post_struct_no_nego(async_app, aiohttp_client) -> None:
"""Test that json encoded posted struct data is returned as msgpack """Test that json encoded posted struct data is returned as msgpack
when using non-negotiation-compatible handlers. when using non-negotiation-compatible handlers.
""" """
cli = await aiohttp_client(async_app) cli = await aiohttp_client(async_app)
for ctype in ('x-msgpack', 'json'): for ctype in ("x-msgpack", "json"):
resp = await cli.post( resp = await cli.post(
'/echo-no-nego', "/echo-no-nego",
headers={'Content-Type': 'application/json', headers={
'Accept': 'application/%s' % ctype}, "Content-Type": "application/json",
data=json.dumps(STRUCT, cls=SWHJSONEncoder)) "Accept": "application/%s" % ctype,
},
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
About · Developer information · Legal