Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/test_rpc_client_server.py
| Show All 14 Lines | class RPCTest: | ||||
| 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') | |||||
| def raise_typeerror(self): | |||||
| raise TypeError('Did I pass through?') | |||||
| # 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') | ||||
| 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('not_on_server') | @remote_api_endpoint('not_on_server') | ||||
| def not_on_server(self, db=None, cur=None): | def not_on_server(self, db=None, cur=None): | ||||
| return 'ok' | return 'ok' | ||||
| @remote_api_endpoint('raises_typeerror') | |||||
| def raise_typeerror(self): | |||||
| return 'data' | |||||
| class RPCTestClient(RPCClient): | class RPCTestClient(RPCClient): | ||||
| backend_class = RPCTest2 | backend_class = RPCTest2 | ||||
| @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 | ||||
| Show All 31 Lines | def test_api_endpoint_kwargs(swh_rpc_client): | ||||
| assert res == 'egg' | assert res == 'egg' | ||||
| def test_api_endpoint_args(swh_rpc_client): | def test_api_endpoint_args(swh_rpc_client): | ||||
| res = swh_rpc_client.something('whatever') | res = swh_rpc_client.something('whatever') | ||||
| assert res == 'whatever' | assert res == 'whatever' | ||||
| res = swh_rpc_client.endpoint('spam') | res = swh_rpc_client.endpoint('spam') | ||||
| assert res == 'egg' | assert res == 'egg' | ||||
| def test_api_typeerror(swh_rpc_client): | |||||
| with pytest.raises(TypeError, match='Did I pass through?'): | |||||
| swh_rpc_client.raise_typeerror() | |||||