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_response_null_id():
# Acceptable.
response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": None})
assert response.ok == True
assert response.id == None
assert repr(response) == ""
assert str(response) == '{"jsonrpc": "2.0", "result": "foo", "id": null}'
def test_success_response():
response = SuccessResponse(**{"jsonrpc": "2.0", "result": 5, "id": 1})
assert response.ok == True
assert response.id == 1
assert response.result == 5
assert repr(response) == ""
assert str(response) == '{"jsonrpc": "2.0", "result": 5, "id": 1}'
def test_total_responses_list():
response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": 1})
assert total_results([response, response]) == 2
def test_total_responses_one():
response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": 1})
assert total_results(response) == 1
ok = True
def __init__(self, result: Any, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.result = result
def __repr__(self) -> str:
return "".format(self.id, self.result)
def __str__(self) -> str:
return serialize(
sort_response(dict(jsonrpc=self.jsonrpc, result=self.result, id=self.id))
)
class NotificationResponse(SuccessResponse):
"""
Represents a JSON-RPC notification response object.
"""
ok = True
def __init__(self) -> None:
super().__init__(jsonrpc="2.0", result=None, id=NOID)
def __repr__(self) -> str:
return ""
def __str__(self) -> str:
return ""
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)
assert isinstance(request, dict),\
'How did you manage to already serialize this request'
method = request['method']
if method in self._funcmap and request.get('params'):
new_params = _apply_arg_transforms(
request['params'],
self._funcmap[method])
LOG.debug(
f"request param xform: {request['params']} -> {new_params}")
request['params'] = new_params
elif method not in self._funcmap:
LOG.warning(
f"jrpc client: no funcmap entry for {request['method']}")
response = await super().send(request, *args, **kwargs)
if isinstance(response.data, SuccessResponse)\
and method in self._funcmap:
# We need to transform the return type
xformed = self._funcmap[method].return_xform(response.data.result)
LOG.debug(f'response xform: {response.data.result}->{xformed}')
response.data.result = xformed
return response