How to use the jsonrpcclient.response.Response function in jsonrpcclient

To help you get started, we’ve selected a few jsonrpcclient examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github bcb / jsonrpcclient / tests / test_client.py View on Github external
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.*'),
            )
github bcb / jsonrpcclient / tests / test_client.py View on Github external
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.*"),
            )
github bcb / jsonrpcclient / tests / test_client.py View on Github external
def send_message(self, request, response_expected):
        return Response('{"jsonrpc": "2.0", "result": 1, "id": 1}')
github bcb / jsonrpcclient / tests / test_client.py View on Github external
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")
github bcb / jsonrpcclient / tests / test_client.py View on Github external
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 + ".*"),
            )
github bcb / jsonrpcclient / tests / test_response.py View on Github external
def test_response_repr():
    response = Response("foo")
    assert repr(response) == ""
github bcb / jsonrpcclient / jsonrpcclient / clients / zeromq_client.py View on Github external
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())
github bcb / jsonrpcclient / jsonrpcclient / clients / socket_client.py View on Github external
# 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])
github bcb / jsonrpcclient / jsonrpcclient / clients / http_client.py View on Github external
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)
github bcb / jsonrpcclient / jsonrpcclient / clients / aiohttp_client.py View on Github external
"""
        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)