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(self):
with LogCapture() as capture:
DummyClient().log_response(
Response('{"jsonrpc": "2.0", "result": 5, "id": 1}')
)
capture.check(
(
"jsonrpcclient.client.response",
"INFO",
StringComparison(r'.*"result": 5.*'),
)
def test_trimmed(self):
req = '{"jsonrpc": "2.0", "result": "%s", "id": 1}' % ("foo" * 100,)
with LogCapture() as capture:
DummyClient().log_response(Response(req), trim_log_values=True)
capture.check(
(
"jsonrpcclient.client.response",
"INFO",
StringComparison(r".*foofoofoof...ofoofoofoo.*"),
)
def send_message(self, request, response_expected):
return Response('{"jsonrpc": "2.0", "result": 1, "id": 1}')
def test_send_single_request_error(*_):
with pytest.raises(ReceivedErrorResponseError):
client = DummyClient()
client.send_message = Mock(
return_value=Response(
'{"jsonrpc": "2.0", "error": {"code": 1, "message": "foo"}, "id": 1}'
)
)
client.request("ping")
def test_untrimmed(self):
"""Should not trim"""
res = '{"jsonrpc": "2.0", "result": {"foo": "%s"}}' % ("foo" * 100,)
with LogCapture() as capture:
DummyClient().log_response(Response(res), trim_log_values=False)
capture.check(
(
"jsonrpcclient.client.response",
"INFO",
StringComparison(r".*" + "foo" * 100 + ".*"),
)
def test_response_repr():
response = Response("foo")
assert repr(response) == ""
def send_message(
self, request: str, response_expected: bool, **kwargs: Any
) -> Response:
"""
Transport the message to the server and return the response.
Args:
request: The JSON-RPC request string.
response_expected: Whether the request expects a response.
Returns:
A Response object.
"""
self.socket.send_string(request)
return Response(self.socket.recv().decode())
# Receive the response until we find the delimiter.
# TODO Do not wait for a response if the message sent is a notification.
while True:
response += self.socket.recv(1024)
decoded = response.decode(self.encoding)
if len(decoded) < self.delimiter_length:
continue
# TODO Check that're not in the middle of the response.
elif decoded[-self.delimiter_length :] == self.delimiter:
break
assert decoded is not None
return Response(decoded[: -self.delimiter_length])
def send_message(
self, request: str, response_expected: bool, **kwargs: Any
) -> Response:
"""
Transport the message to the server and return the response.
Args:
request: The JSON-RPC request string.
response_expected: Whether the request expects a response.
Returns:
A Response object.
"""
response = self.session.post(self.endpoint, data=request.encode(), **kwargs)
return Response(response.text, raw=response)
"""
Transport the message to the server and return the response.
Args:
request: The JSON-RPC request string.
response_expected: Whether the request expects a response.
Returns:
A Response object.
"""
with async_timeout.timeout(self.timeout):
async with self.session.post(
self.endpoint, data=request, ssl=self.ssl, **kwargs
) as response:
response_text = await response.text()
return Response(response_text, raw=response)