Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
(hvac.exceptions.VaultDown, exceptions.VaultSealed),
(hvac.exceptions.UnexpectedError, exceptions.VaultAPIException),
],
)
def test_handle_errors(hvac_exc, vault_cli_exc):
with pytest.raises(vault_cli_exc):
with client.handle_errors():
raise hvac_exc
tenacity.retry_if_exception(hvac.exceptions.VaultDown)
)
)
def _do_it():
client = _vault_client(config)
func(args, client, config)
_do_it()
except json.decoder.JSONDecodeError as exc:
raise exceptions.VaultNonJsonResponse(errors=[str(exc)])
except hvac.exceptions.InvalidRequest as exc:
raise exceptions.VaultInvalidRequest(errors=exc.errors) from exc
except hvac.exceptions.Unauthorized as exc:
raise exceptions.VaultUnauthorized(errors=exc.errors) from exc
except hvac.exceptions.Forbidden as exc:
raise exceptions.VaultForbidden(errors=exc.errors) from exc
except hvac.exceptions.InternalServerError as exc:
raise exceptions.VaultInternalServerError(errors=exc.errors) from exc
except hvac.exceptions.VaultDown as exc:
raise exceptions.VaultSealed(errors=exc.errors) from exc
except hvac.exceptions.UnexpectedError as exc:
raise exceptions.VaultAPIException(errors=exc.errors) from exc
if status_code == 400:
raise exceptions.InvalidRequest(message, errors=errors)
elif status_code == 401:
raise exceptions.Unauthorized(message, errors=errors)
elif status_code == 403:
raise exceptions.Forbidden(message, errors=errors)
elif status_code == 404:
raise exceptions.InvalidPath(message, errors=errors)
elif status_code == 429:
raise exceptions.RateLimitExceeded(message, errors=errors)
elif status_code == 500:
raise exceptions.InternalServerError(message, errors=errors)
elif status_code == 501:
raise exceptions.VaultNotInitialized(message, errors=errors)
elif status_code == 503:
raise exceptions.VaultDown(message, errors=errors)
else:
raise exceptions.UnexpectedError(message)
if status_code == 400:
raise exceptions.InvalidRequest(message, errors=errors)
elif status_code == 401:
raise exceptions.Unauthorized(message, errors=errors)
elif status_code == 403:
raise exceptions.Forbidden(message, errors=errors)
elif status_code == 404:
raise exceptions.InvalidPath(message, errors=errors)
elif status_code == 429:
raise exceptions.RateLimitExceeded(message, errors=errors)
elif status_code == 500:
raise exceptions.InternalServerError(message, errors=errors)
elif status_code == 501:
raise exceptions.VaultNotInitialized(message, errors=errors)
elif status_code == 503:
raise exceptions.VaultDown(message, errors=errors)
else:
raise exceptions.UnexpectedError(message)
def handle_errors():
try:
yield
except json.decoder.JSONDecodeError as exc:
raise exceptions.VaultNonJsonResponse(errors=[str(exc)])
except hvac.exceptions.InvalidRequest as exc:
raise exceptions.VaultInvalidRequest(errors=exc.errors) from exc
except hvac.exceptions.Unauthorized as exc:
raise exceptions.VaultUnauthorized(errors=exc.errors) from exc
except hvac.exceptions.Forbidden as exc:
raise exceptions.VaultForbidden(errors=exc.errors) from exc
except hvac.exceptions.InternalServerError as exc:
raise exceptions.VaultInternalServerError(errors=exc.errors) from exc
except hvac.exceptions.VaultDown as exc:
raise exceptions.VaultSealed(errors=exc.errors) from exc
except hvac.exceptions.UnexpectedError as exc:
raise exceptions.VaultAPIException(errors=exc.errors) from exc