Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_success(self):
req = Request({"jsonrpc": "2.0", "method": "foo"}).call([foo])
self.assertIsInstance(req, NotificationResponse)
def test_method_not_found(self):
req = Request({"jsonrpc": "2.0", "method": "baz"}).call([foo])
self.assertIsInstance(req, NotificationResponse)
def test_dispatch_pure_notification():
response = dispatch_pure(
'{"jsonrpc": "2.0", "method": "ping"}',
Methods(ping),
convert_camel_case=False,
context=NOCONTEXT,
debug=True,
serialize=default_serialize,
deserialize=default_deserialize,
)
assert isinstance(response, NotificationResponse)
def test_safe_call_notification():
response = safe_call(
Request(method="ping"), Methods(ping), debug=True, serialize=default_serialize
)
assert isinstance(response, NotificationResponse)
def test_notification_response_str():
assert str(NotificationResponse()) == ""
# InvalidParamsError is raised by validate_args. AssertionError is raised inside
# the methods, however it's better to raise InvalidParamsError inside methods.
# AssertionError will be removed in the next major release.
handler.response = InvalidParamsResponse(
id=request.id, data=str(exc), debug=debug
)
except ApiError as exc: # Method signals custom error
handler.response = ApiErrorResponse(
str(exc), code=exc.code, data=exc.data, id=request.id, debug=debug
)
except Exception as exc: # Other error inside method - server error
logging.exception(exc)
handler.response = ExceptionResponse(exc, id=request.id, debug=debug)
finally:
if request.is_notification:
handler.response = NotificationResponse()
async def call(self, methods):
# Validation or parsing may have failed in __init__, in which case
# there's no point calling. It would've already set the response.
if not self.response:
# Handles setting the result/exception of the call
with self.handle_exceptions():
# Get the method object from a list (raises MethodNotFound)
callable_ = self.get_method(methods)
# Ensure the arguments match the method's signature
validate_arguments_against_signature(callable_, self.args, self.kwargs)
# Call the method
result = await callable_(*(self.args or []), **(self.kwargs or {}))
# Set the response
if self.is_notification:
self.response = NotificationResponse()
else:
self.response = RequestResponse(self.request_id, result)
# Ensure the response has been set before returning it
assert isinstance(self.response, Response), "Invalid response type"
return self.response