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_failed_download(http_client, http_server):
"""Tests that a failed download raises an HTTPError."""
http_server.serve_content("fail", 500)
with pytest.raises(ipfshttpclient.exceptions.StatusError):
http_client.download("/fail")
def fetchRemoteCDXJFile(path):
fileContents = ''
path = path.replace('ipfs://', '')
# TODO: Take into account /ipfs/(hash), first check if this is correct fmt
if '://' not in path: # isAIPFSHash
# TODO: Check if a valid IPFS hash
print('No scheme in path, assuming IPFS hash and fetching...')
try:
print("Trying to ipfs.cat('{0}')".format(path))
dataFromIPFS = IPFS_API.cat(path)
except hashNotInIPFS:
print(("The CDXJ at hash {0} could"
" not be found in IPFS").format(path))
sys.exit()
except Exception as e:
print("An error occurred with ipfs.cat")
print(sys.exc_info()[0])
sys.exit()
print('Data successfully obtained from IPFS')
return dataFromIPFS
else: # http://, ftp://, smb://, file://
print('Path contains a scheme, fetching remote file...')
fileContents = ipwbUtils.fetchRemoteFile(path)
return fileContents
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)
msg = "{}: {}".format(original.__class__.__name__, str(original))
Error.__init__(self, msg)
class ProtocolError(CommunicationError):
"""Raised when parsing the response from the daemon has failed.
This can most likely occur if the service on the remote end isn't in fact
an IPFS daemon."""
class StatusError(CommunicationError):
"""Raised when the daemon responds with an error to our request."""
class ErrorResponse(StatusError):
"""Raised when the daemon has responded with an error message because the
requested operation could not be carried out."""
def __init__(self, message, original):
StatusError.__init__(self, original, message)
class PartialErrorResponse(ErrorResponse):
"""Raised when the daemon has responded with an error message after having
already returned some data.
The incomplete data returned may be accessed using the ``partial``
attribute."""
def __init__(self, message, original, partial):
self.partial = partial
def __init__(self, message, original):
StatusError.__init__(self, original, message)