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_error_response_no_id():
deserialized = {"jsonrpc": "2.0", "error": {"code": -32000, "message": "Not Found"}}
response = ErrorResponse(**deserialized)
assert repr(response) == ''
assert (
str(response)
== '{"jsonrpc": "2.0", "error": {"code": -32000, "message": "Not Found"}}'
)
def test_response_repr_with_results():
response = Response("foo")
response.data = ErrorResponse(
**{"jsonrpc": "2.0", "error": {"message": "foo"}, "id": 1}
)
assert repr(response) == ""
def test_error_response():
deserialized = {
"jsonrpc": "2.0",
"error": {"code": -32000, "message": "Not Found", "data": "foo"},
"id": 1,
}
response = ErrorResponse(**deserialized)
assert response.ok == False
assert response.id == 1
assert response.code == -32000
assert response.message == "Not Found"
assert response.data == "foo"
assert repr(response) == ''
assert (
str(response)
== '{"jsonrpc": "2.0", "error": {"code": -32000, "message": "Not Found", "data": "foo"}, "id": 1}'
)
def test_error_with_data():
response = {
"jsonrpc": "2.0",
"error": {
"code": -32000,
"message": "Not Found",
"data": "Lorem ipsum dolor sit amet, consectetur adipiscing elit",
},
"id": None,
}
response = ErrorResponse(**response)
assert response.code == -32000
assert response.message == "Not Found"
assert response.data == "Lorem ipsum dolor sit amet, consectetur adipiscing elit"
def test_error_with_nonstring_data():
"""Reported in issue #56"""
response = {
"jsonrpc": "2.0",
"error": {"code": -32000, "message": "Not Found", "data": {}},
"id": None,
}
response = ErrorResponse(**response)
assert response.code == -32000
assert response.message == "Not Found"
assert response.data == {}
else:
request_text = serialize(request)
request_deserialized = request
batch = isinstance(request_deserialized, list)
response_expected = batch or "id" in request_deserialized
self.log_request(request_text, trim_log_values=trim_log_values)
response = await self.send_message(
request_text, response_expected=response_expected, **kwargs
)
self.log_response(response, trim_log_values=trim_log_values)
self.validate_response(response)
response.data = parse(
response.text, batch=batch, validate_against_schema=validate_against_schema
)
# If received a single error response, raise
if isinstance(response.data, ErrorResponse):
raise ReceivedErrorResponseError(response.data)
return response
def get_response(response: Dict[str, Any]) -> JSONRPCResponse:
"""
Converts a deserialized response into a JSONRPCResponse object.
The dictionary be either an error or success response, never a notification.
Args:
response: Deserialized response dictionary. We can assume the response is valid
JSON-RPC here, since it passed the jsonschema validation.
"""
if "error" in response:
return ErrorResponse(**response)
return SuccessResponse(**response)
else:
request_text = serialize(request)
request_deserialized = request
batch = isinstance(request_deserialized, list)
response_expected = batch or "id" in request_deserialized
self.log_request(request_text, trim_log_values=trim_log_values)
response = self.send_message(
request_text, response_expected=response_expected, **kwargs
)
self.log_response(response, trim_log_values=trim_log_values)
self.validate_response(response)
response.data = parse(
response.text, batch=batch, validate_against_schema=validate_against_schema
)
# If received a single error response, raise
if isinstance(response.data, ErrorResponse):
raise ReceivedErrorResponseError(response.data)
return response