Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_async.py
| Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines | def async_app(): | ||||
| 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: | @pytest.fixture | ||||
| assert async_app is not None | def cli(async_app, aiohttp_client, loop): | ||||
| return loop.run_until_complete(aiohttp_client(async_app)) | |||||
| cli = await aiohttp_client(async_app) | async def test_get_simple(cli) -> None: | ||||
| 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(cli) -> None: | ||||
| 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(cli) -> None: | ||||
| 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(cli) -> None: | ||||
| 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(cli) -> 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) | |||||
| 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(cli) -> 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) | |||||
| for ctype in ("x-msgpack", "json"): | for ctype in ("x-msgpack", "json"): | ||||
| resp = await cli.get("/struct", headers={"Accept": "application/%s" % ctype}) | resp = await cli.get("/struct", 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(cli) -> 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) | |||||
| # 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(cli) -> 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) | |||||
| 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(cli) -> 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) | |||||
| for ctype in ("x-msgpack", "json"): | for ctype in ("x-msgpack", "json"): | ||||
| resp = await cli.post( | resp = await cli.post( | ||||
| "/echo", | "/echo", | ||||
| headers={ | headers={ | ||||
| "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/%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(cli) -> 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) | |||||
| 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={ | headers={ | ||||
| "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 | ||||