Differential D8227 Diff 29675 swh/core/api/tests/test_rpc_client_server.py

Changeset View

Standalone View

swh/core/api/tests/test_rpc_client_server.py

# Copyright (C) 2018-2019 The Software Heritage developers		# Copyright (C) 2018-2022 The Software Heritage developers
# 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 pytest		import pytest

from swh.core.api import (		from swh.core.api import (
RemoteException,		RemoteException,
RPCClient,		RPCClient,
RPCServerApp,		RPCServerApp,
encode_data_server,		encode_data_server,
error_handler,		error_handler,
remote_api_endpoint,		remote_api_endpoint,
)		)


class ExpectedException(Exception):		class ExpectedException(Exception):
"""Another exception class to distinguish error handlers"""		"""Another exception class to distinguish error handlers"""


		class NotFoundException(Exception):
		pass


# this class is used on the server part		# this class is used on the server part
class RPCTest:		class RPCTest:
@remote_api_endpoint("endpoint_url")		@remote_api_endpoint("endpoint_url")
def endpoint(self, test_data, db=None, cur=None):		def 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("raises_typeerror")		@remote_api_endpoint("raises_typeerror")
def raise_typeerror(self):		def raise_typeerror(self):
raise TypeError("Did I pass through?")		raise TypeError("Did I pass through?")

@remote_api_endpoint("raise_exception_exc_arg")		@remote_api_endpoint("raise_exception_exc_arg")
def raise_exception_exc_arg(self):		def raise_exception_exc_arg(self):
raise Exception(Exception("error"))		raise Exception(Exception("error"))

@remote_api_endpoint("raises_expectedexc")		@remote_api_endpoint("raises_expectedexc")
def raise_expectedexc(self):		def raise_expectedexc(self):
raise ExpectedException("that was expected")		raise ExpectedException("that was expected")

		@remote_api_endpoint("raises_not_found_exc")
		def raise_not_found_exc(self):
		raise NotFoundException()


# this class is used on the client part. We cannot inherit from RPCTest		# this class is used on the client part. We cannot inherit from RPCTest
# because the automagic metaclass based code that generates the RPCClient		# because the automagic metaclass based code that generates the RPCClient
# proxy class from this does not handle inheritance properly.		# proxy class from this does not handle inheritance properly.
# We do add an endpoint on the client side that has no implementation		# We do add an endpoint on the client side that has no implementation
# server-side to test this very situation (in should generate a 404)		# server-side to test this very situation (in should generate a 404)
class RPCTest2:		class RPCTest2:
@remote_api_endpoint("endpoint_url")		@remote_api_endpoint("endpoint_url")
Show All 12 Lines	class RPCTest2:
@remote_api_endpoint("raises_typeerror")		@remote_api_endpoint("raises_typeerror")
def raise_typeerror(self):		def raise_typeerror(self):
return "data"		return "data"

@remote_api_endpoint("raises_expectedexc")		@remote_api_endpoint("raises_expectedexc")
def raise_expectedexc(self):		def raise_expectedexc(self):
return "nothing"		return "nothing"

		@remote_api_endpoint("raises_not_found_exc")
		def raise_not_found_exc(self):
		return "not found"


class RPCTestClient(RPCClient):		class RPCTestClient(RPCClient):
backend_class = RPCTest2		backend_class = RPCTest2
		reraise_exceptions = [NotFoundException]


@pytest.fixture		@pytest.fixture
def app():		def app():
# This fixture is used by the 'swh_rpc_adapter' fixture		# This fixture is used by the 'swh_rpc_adapter' fixture
# which is defined in swh/core/pytest_plugin.py		# which is defined in swh/core/pytest_plugin.py
application = RPCServerApp("testapp", backend_class=RPCTest)		application = RPCServerApp("testapp", backend_class=RPCTest)

@application.errorhandler(ExpectedException)		@application.errorhandler(ExpectedException)
def my_expected_error_handler(exception):		def my_expected_error_handler(exception):
return error_handler(exception, encode_data_server, status_code=400)		return error_handler(exception, encode_data_server, status_code=400)

		@application.errorhandler(NotFoundException)
		def not_found_error_handler(exception):
		return error_handler(exception, encode_data_server, status_code=404)

@application.errorhandler(Exception)		@application.errorhandler(Exception)
def my_error_handler(exception):		def my_error_handler(exception):
return error_handler(exception, encode_data_server)		return error_handler(exception, encode_data_server)

return application		return application


@pytest.fixture		@pytest.fixture
Show All 10 Lines

def test_api_server_endpoint_missing(swh_rpc_client):		def test_api_server_endpoint_missing(swh_rpc_client):
# A 'missing' endpoint (server-side) should raise an exception		# A 'missing' endpoint (server-side) should raise an exception
# due to a 404, since at the end, we do a GET/POST an inexistent URL		# due to a 404, since at the end, we do a GET/POST an inexistent URL
with pytest.raises(Exception, match="404 not found"):		with pytest.raises(Exception, match="404 not found"):
swh_rpc_client.not_on_server()		swh_rpc_client.not_on_server()


		def test_api_reraise_not_found(swh_rpc_client):
		# A 'missing' endpoint (server-side) should raise an exception
		# due to a 404, since at the end, we do a GET/POST an inexistent URL
		with pytest.raises(NotFoundException):
		swh_rpc_client.raise_not_found_exc()


def test_api_endpoint_kwargs(swh_rpc_client):		def test_api_endpoint_kwargs(swh_rpc_client):
res = swh_rpc_client.something(data="whatever")		res = swh_rpc_client.something(data="whatever")
assert res == "whatever"		assert res == "whatever"
res = swh_rpc_client.endpoint(test_data="spam")		res = swh_rpc_client.endpoint(test_data="spam")
assert res == "egg"		assert res == "egg"


def test_api_endpoint_args(swh_rpc_client):		def test_api_endpoint_args(swh_rpc_client):
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines