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_get_encoder_by_name():
"""Tests the process of obtaining an Encoder object given the named encoding."""
encoder = ipfshttpclient.encoding.get_encoding('json')
assert encoder.name == 'json'
def test_get_invalid_encoder():
"""Tests the exception handling given an invalid named encoding."""
with pytest.raises(ipfshttpclient.exceptions.EncoderMissingError):
ipfshttpclient.encoding.get_encoding('fake')
def _do_raise_for_status(self, response):
try:
response.raise_for_status()
except requests.exceptions.HTTPError as error:
content = []
try:
decoder = encoding.get_encoding("json")
for chunk in response.iter_content(chunk_size=None):
content += list(decoder.parse_partial(chunk))
content += list(decoder.parse_finalize())
except exceptions.DecodingError:
pass
# If we have decoded an error response from the server,
# use that as the exception message; otherwise, just pass
# the exception on to the caller.
if len(content) == 1 \
and isinstance(content[0], dict) \
and "Message" in content[0]:
msg = content[0]["Message"]
six.raise_from(exceptions.ErrorResponse(msg, error), error)
else:
six.raise_from(exceptions.StatusError(error), error)
params.append(('arg', arg))
if (files or data):
method = 'post'
elif not return_result:
method = 'head'
else:
method = 'get'
# Don't attempt to decode response or stream
# (which would keep an iterator open that will then never be waited for)
if not return_result:
decoder = None
stream = False
parser = encoding.get_encoding(decoder if decoder else "none")
ret = self._request(method, url, params, parser, stream,
files, headers, data, timeout=timeout)
return ret if return_result else None