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_parse_invalid_jsonrpc():
with pytest.raises(ValidationError):
parse('{"json": "2.0"}', batch=False)
def test_parse_empty_string_single():
assert parse("", batch=False).result is None
def test_parse_none_single():
assert parse(None, batch=False).result is None
def test_parse_without_validation():
parse(
'{"jsonrpc": "2.0", "result": "foo", "id": 1}',
batch=False,
validate_against_schema=False,
)
def test_parse_empty_string_batch():
assert parse("", batch=True) == []
# We need both the serialized and deserialized version of the request
if isinstance(request, str):
request_text = request
request_deserialized = deserialize(request)
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
# We need both the serialized and deserialized version of the request
if isinstance(request, str):
request_text = request
request_deserialized = deserialize(request)
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