Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_async.py
| Show First 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | |||||
| 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 app(): | def async_app(): | ||||
| app = RPCServerApp() | app = RPCServerApp() | ||||
| 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('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(app, aiohttp_client) -> None: | async def test_get_simple(async_app, aiohttp_client) -> None: | ||||
| assert app is not None | assert async_app is not None | ||||
| cli = await aiohttp_client(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_simple_nego(app, aiohttp_client) -> None: | async def test_get_simple_nego(async_app, aiohttp_client) -> None: | ||||
| cli = await aiohttp_client(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(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(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(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(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(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(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(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(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(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(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={'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(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(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={'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 | ||||