Differential D8220 Diff 29655 swh/core/api/tests/test_rpc_server.py

Changeset View

Standalone View

swh/core/api/tests/test_rpc_server.py

# Copyright (C) 2018-2019 The Software Heritage developers		# Copyright (C) 2018-2019 The Software Heritage developers
		anlambertUnsubmitted Not Done Inline Actions 2022 anlambert: 2022
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import json		import json

from flask import url_for		from flask import url_for
import msgpack		import msgpack
import pytest		import pytest

from swh.core.api import (		from swh.core.api import (
JSONFormatter,		JSONFormatter,
MsgpackFormatter,		MsgpackFormatter,
RPCServerApp,		RPCServerApp,
		encode_data_server,
		error_handler,
negotiate,		negotiate,
remote_api_endpoint,		remote_api_endpoint,
)		)

from .test_serializers import ExtraType, extra_decoders, extra_encoders		from .test_serializers import ExtraType, extra_decoders, extra_encoders


		class MyCustomException(Exception):
		pass


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


class TestStorage:		class TestStorage:
@remote_api_endpoint("test_endpoint_url")		@remote_api_endpoint("test_endpoint_url")
def endpoint_test(self, test_data, db=None, cur=None):		def endpoint_test(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")

		@remote_api_endpoint("crashy/builtin")
		def crashy(self, data, db=None, cur=None):
		raise ValueError("this is an unexpected exception")

		@remote_api_endpoint("crashy/custom")
		def custom_crashy(self, data, db=None, cur=None):
		raise MyCustomException("try again later!")


@pytest.fixture		@pytest.fixture
def app():		def app():
return MyRPCServerApp("testapp", backend_class=TestStorage)		app = MyRPCServerApp("testapp", backend_class=TestStorage)

		@app.errorhandler(MyCustomException)
		def custom_error_handler(exception):
		return error_handler(exception, encode_data_server, status_code=503)

		@app.errorhandler(Exception)
		def default_error_handler(exception):
		return error_handler(exception, encode_data_server)

		return app


def test_api_rpc_server_app_ok(app):		def test_api_rpc_server_app_ok(app):
assert isinstance(app, MyRPCServerApp)		assert isinstance(app, MyRPCServerApp)

actual_rpc_server2 = MyRPCServerApp(		actual_rpc_server2 = MyRPCServerApp(
"app2", backend_class=TestStorage, backend_factory=TestStorage		"app2", backend_class=TestStorage, backend_factory=TestStorage
)		)
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines	res = flask_app_client.post(
data=b"\x81\xa9test_data\xa4spam",		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_exception(flask_app_client):
		res = flask_app_client.post(
		url_for("crashy"),
		headers=[
		("Content-Type", "application/x-msgpack"),
		("Accept", "application/x-msgpack"),
		],
		data=msgpack.dumps({"data": "toto"}),
		)

		assert res.status_code == 500
		assert res.mimetype == "application/x-msgpack", res.data
		data = msgpack.loads(res.data)
		assert (
		data.items()
		>= {
		"type": "ValueError",
		"module": "builtins",
		"args": ["this is an unexpected exception"],
		}.items()
		), data


		def test_rpc_server_custom_exception(flask_app_client):
		res = flask_app_client.post(
		url_for("custom_crashy"),
		headers=[
		("Content-Type", "application/x-msgpack"),
		("Accept", "application/x-msgpack"),
		],
		data=msgpack.dumps({"data": "toto"}),
		)

		assert res.status_code == 503
		assert res.mimetype == "application/x-msgpack", res.data
		data = msgpack.loads(res.data)
		assert (
		data.items()
		>= {
		"type": "MyCustomException",
		"module": "swh.core.api.tests.test_rpc_server",
		"args": ["try again later!"],
		}.items()
		), data


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=[		headers=[
("Content-Type", "application/x-msgpack"),		("Content-Type", "application/x-msgpack"),
("Accept", "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"
Show All 26 Lines