Use specific exception to prevent repetition
This commit is contained in:
parent
8eba88d1af
commit
4eb921480e
@ -15,6 +15,15 @@ class ResourceNotFoundException(SecretsManagerClientError):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class SecretNotFoundException(SecretsManagerClientError):
|
||||||
|
def __init__(self):
|
||||||
|
self.code = 404
|
||||||
|
super(SecretNotFoundException, self).__init__(
|
||||||
|
"ResourceNotFoundException",
|
||||||
|
message=u"Secrets Manager can\u2019t find the specified secret."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class ClientError(SecretsManagerClientError):
|
class ClientError(SecretsManagerClientError):
|
||||||
def __init__(self, message):
|
def __init__(self, message):
|
||||||
super(ClientError, self).__init__(
|
super(ClientError, self).__init__(
|
||||||
|
@ -11,6 +11,7 @@ import boto3
|
|||||||
from moto.core import BaseBackend, BaseModel
|
from moto.core import BaseBackend, BaseModel
|
||||||
from .exceptions import (
|
from .exceptions import (
|
||||||
ResourceNotFoundException,
|
ResourceNotFoundException,
|
||||||
|
SecretNotFoundException,
|
||||||
InvalidParameterException,
|
InvalidParameterException,
|
||||||
ResourceExistsException,
|
ResourceExistsException,
|
||||||
InvalidRequestException,
|
InvalidRequestException,
|
||||||
@ -47,7 +48,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def get_secret_value(self, secret_id, version_id, version_stage):
|
def get_secret_value(self, secret_id, version_id, version_stage):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
if not version_id and version_stage:
|
if not version_id and version_stage:
|
||||||
# set version_id to match version_stage
|
# set version_id to match version_stage
|
||||||
@ -57,7 +58,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
version_id = ver_id
|
version_id = ver_id
|
||||||
break
|
break
|
||||||
if not version_id:
|
if not version_id:
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
# TODO check this part
|
# TODO check this part
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
@ -176,7 +177,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
|
|
||||||
def describe_secret(self, secret_id):
|
def describe_secret(self, secret_id):
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
secret = self.secrets[secret_id]
|
secret = self.secrets[secret_id]
|
||||||
|
|
||||||
@ -205,7 +206,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
rotation_days = 'AutomaticallyAfterDays'
|
rotation_days = 'AutomaticallyAfterDays'
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
raise InvalidRequestException(
|
raise InvalidRequestException(
|
||||||
@ -347,7 +348,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def delete_secret(self, secret_id, recovery_window_in_days, force_delete_without_recovery):
|
def delete_secret(self, secret_id, recovery_window_in_days, force_delete_without_recovery):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
if 'deleted_date' in self.secrets[secret_id]:
|
if 'deleted_date' in self.secrets[secret_id]:
|
||||||
raise InvalidRequestException(
|
raise InvalidRequestException(
|
||||||
@ -377,7 +378,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
secret = self.secrets.get(secret_id, None)
|
secret = self.secrets.get(secret_id, None)
|
||||||
|
|
||||||
if not secret:
|
if not secret:
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
arn = secret_arn(self.region, secret['secret_id'])
|
arn = secret_arn(self.region, secret['secret_id'])
|
||||||
name = secret['name']
|
name = secret['name']
|
||||||
@ -387,7 +388,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
def restore_secret(self, secret_id):
|
def restore_secret(self, secret_id):
|
||||||
|
|
||||||
if not self._is_valid_identifier(secret_id):
|
if not self._is_valid_identifier(secret_id):
|
||||||
raise ResourceNotFoundException(u"Secrets Manager can’t find the specified secret.")
|
raise SecretNotFoundException()
|
||||||
|
|
||||||
self.secrets[secret_id].pop('deleted_date', None)
|
self.secrets[secret_id].pop('deleted_date', None)
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user