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"' | ||||