Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_server.py
Show All 16 Lines | |||||
class MyRPCServerApp(RPCServerApp): | class MyRPCServerApp(RPCServerApp): | ||||
extra_type_encoders = extra_encoders | extra_type_encoders = extra_encoders | ||||
extra_type_decoders = extra_decoders | extra_type_decoders = extra_decoders | ||||
@pytest.fixture | @pytest.fixture | ||||
def app(): | def app(): | ||||
class TestStorage: | class TestStorage: | ||||
@remote_api_endpoint('test_endpoint_url') | @remote_api_endpoint("test_endpoint_url") | ||||
def test_endpoint(self, test_data, db=None, cur=None): | def test_endpoint(self, test_data, db=None, cur=None): | ||||
assert test_data == 'spam' | assert test_data == "spam" | ||||
return 'egg' | return "egg" | ||||
@remote_api_endpoint('path/to/endpoint') | @remote_api_endpoint("path/to/endpoint") | ||||
def something(self, data, db=None, cur=None): | def something(self, data, db=None, cur=None): | ||||
return data | return data | ||||
@remote_api_endpoint('serializer_test') | @remote_api_endpoint("serializer_test") | ||||
def serializer_test(self, data, db=None, cur=None): | def serializer_test(self, data, db=None, cur=None): | ||||
assert data == ['foo', ExtraType('bar', b'baz')] | assert data == ["foo", ExtraType("bar", b"baz")] | ||||
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"), | ||||
data=msgpack.dumps({'data': 'toto'}), | ("Content-Type", "application/x-msgpack"), | ||||
], | |||||
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"), | ||||
data=b'\x81\xa4data\x92\xa3foo\x82\xc4\x07swhtype\xa9extratype' | ("Accept", "application/x-msgpack"), | ||||
b'\xc4\x01d\x92\xa3bar\xc4\x03baz') | ], | ||||
data=b"\x81\xa4data\x92\xa3foo\x82\xc4\x07swhtype\xa9extratype" | |||||
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"' |