Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_server.py
Show All 36 Lines | class TestStorage: | ||||
return ExtraType({'spam': 'egg'}, 'qux') | return ExtraType({'spam': 'egg'}, 'qux') | ||||
return MyRPCServerApp('testapp', backend_class=TestStorage) | return MyRPCServerApp('testapp', backend_class=TestStorage) | ||||
def test_api_endpoint(flask_app_client): | def test_api_endpoint(flask_app_client): | ||||
res = flask_app_client.post( | res = flask_app_client.post( | ||||
url_for('something'), | url_for('something'), | ||||
headers=[('Content-Type', 'application/json'), | headers=[('Content-Type', 'application/json'), ('Accept', 'application/json')], | ||||
('Accept', 'application/json')], | |||||
data=json.dumps({'data': 'toto'}), | data=json.dumps({'data': 'toto'}), | ||||
) | ) | ||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/json' | assert res.mimetype == 'application/json' | ||||
def test_api_nego_default(flask_app_client): | def test_api_nego_default(flask_app_client): | ||||
res = flask_app_client.post( | res = flask_app_client.post( | ||||
url_for('something'), | url_for('something'), | ||||
headers=[('Content-Type', 'application/json')], | headers=[('Content-Type', 'application/json')], | ||||
data=json.dumps({'data': 'toto'}), | data=json.dumps({'data': 'toto'}), | ||||
) | ) | ||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/json' | assert res.mimetype == 'application/json' | ||||
assert res.data == b'"toto"' | assert res.data == b'"toto"' | ||||
def test_api_nego_accept(flask_app_client): | def test_api_nego_accept(flask_app_client): | ||||
res = flask_app_client.post( | res = flask_app_client.post( | ||||
url_for('something'), | url_for('something'), | ||||
headers=[('Accept', 'application/x-msgpack'), | headers=[ | ||||
('Content-Type', 'application/x-msgpack')], | ('Accept', 'application/x-msgpack'), | ||||
('Content-Type', 'application/x-msgpack'), | |||||
], | |||||
data=msgpack.dumps({'data': 'toto'}), | data=msgpack.dumps({'data': 'toto'}), | ||||
) | ) | ||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/x-msgpack' | assert res.mimetype == 'application/x-msgpack' | ||||
assert res.data == b'\xa4toto' | assert res.data == b'\xa4toto' | ||||
def test_rpc_server(flask_app_client): | def test_rpc_server(flask_app_client): | ||||
res = flask_app_client.post( | res = flask_app_client.post( | ||||
url_for('test_endpoint'), | url_for('test_endpoint'), | ||||
headers=[('Content-Type', 'application/x-msgpack'), | headers=[ | ||||
('Accept', 'application/x-msgpack')], | ('Content-Type', 'application/x-msgpack'), | ||||
data=b'\x81\xa9test_data\xa4spam') | ('Accept', 'application/x-msgpack'), | ||||
], | |||||
data=b'\x81\xa9test_data\xa4spam', | |||||
) | |||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/x-msgpack' | assert res.mimetype == 'application/x-msgpack' | ||||
assert res.data == b'\xa3egg' | assert res.data == b'\xa3egg' | ||||
def test_rpc_server_extra_serializers(flask_app_client): | def test_rpc_server_extra_serializers(flask_app_client): | ||||
res = flask_app_client.post( | res = flask_app_client.post( | ||||
url_for('serializer_test'), | url_for('serializer_test'), | ||||
headers=[('Content-Type', 'application/x-msgpack'), | headers=[ | ||||
('Accept', 'application/x-msgpack')], | ('Content-Type', 'application/x-msgpack'), | ||||
('Accept', 'application/x-msgpack'), | |||||
], | |||||
data=b'\x81\xa4data\x92\xa3foo\x82\xc4\x07swhtype\xa9extratype' | data=b'\x81\xa4data\x92\xa3foo\x82\xc4\x07swhtype\xa9extratype' | ||||
b'\xc4\x01d\x92\xa3bar\xc4\x03baz') | b'\xc4\x01d\x92\xa3bar\xc4\x03baz', | ||||
) | |||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/x-msgpack' | assert res.mimetype == 'application/x-msgpack' | ||||
assert res.data == ( | assert res.data == ( | ||||
b'\x82\xc4\x07swhtype\xa9extratype\xc4' | b'\x82\xc4\x07swhtype\xa9extratype\xc4' b'\x01d\x92\x81\xa4spam\xa3egg\xa3qux' | ||||
b'\x01d\x92\x81\xa4spam\xa3egg\xa3qux') | ) | ||||
def test_api_negotiate_no_extra_encoders(app, flask_app_client): | def test_api_negotiate_no_extra_encoders(app, flask_app_client): | ||||
url = '/test/negotiate/no/extra/encoders' | url = '/test/negotiate/no/extra/encoders' | ||||
@app.route(url, methods=['POST']) | @app.route(url, methods=['POST']) | ||||
@negotiate(MsgpackFormatter) | @negotiate(MsgpackFormatter) | ||||
@negotiate(JSONFormatter) | @negotiate(JSONFormatter) | ||||
def endpoint(): | def endpoint(): | ||||
return 'test' | return 'test' | ||||
res = flask_app_client.post( | res = flask_app_client.post(url, headers=[('Content-Type', 'application/json')]) | ||||
url, | |||||
headers=[('Content-Type', 'application/json')], | |||||
) | |||||
assert res.status_code == 200 | assert res.status_code == 200 | ||||
assert res.mimetype == 'application/json' | assert res.mimetype == 'application/json' | ||||
assert res.data == b'"test"' | assert res.data == b'"test"' |