add DeletionProtectionEnabled to dynamoDB (#7619)

moto/dynamodb/exceptions.py

@@ -381,3 +381,9 @@ class UnknownKeyType(MockValidationException):
     def __init__(self, key_type: str, position: str):
         msg = f"1 validation error detected: Value '{key_type}' at '{position}' failed to satisfy constraint: Member must satisfy enum value set: [HASH, RANGE]"
         super().__init__(msg)
+
+
+class DeletionProtectedException(MockValidationException):
+    def __init__(self, table_name: str):
+        msg = f"1 validation error detected: Table '{table_name}' can't be deleted while DeletionProtectionEnabled is set to True"
+        super().__init__(msg)

moto/dynamodb/models/__init__.py

@@ -10,6 +10,7 @@
 from moto.dynamodb.exceptions import (
     BackupNotFoundException,
     ConditionalCheckFailed,
+    DeletionProtectedException,
     ItemSizeTooLarge,
     ItemSizeToUpdateTooLarge,
     MockValidationException,
@@ -71,6 +72,10 @@ def create_table(self, name: str, **params: Any) -> Table:
     def delete_table(self, name: str) -> Table:
         if name not in self.tables:
             raise ResourceNotFoundException
+        table_for_deletion = self.tables.get(name)
+        if isinstance(table_for_deletion, Table):
+            if table_for_deletion.deletion_protection_enabled:
+                raise DeletionProtectedException(name)
         return self.tables.pop(name)
 
     def describe_endpoints(self) -> List[Dict[str, Union[int, str]]]:
@@ -137,6 +142,7 @@ def update_table(
         throughput: Dict[str, Any],
         billing_mode: str,
         stream_spec: Dict[str, Any],
+        deletion_protection_enabled: bool,
     ) -> Table:
         table = self.get_table(name)
         if attr_definitions:
@@ -149,6 +155,10 @@ def update_table(
             table = self.update_table_billing_mode(name, billing_mode)
         if stream_spec:
             table = self.update_table_streams(name, stream_spec)
+        if deletion_protection_enabled:
+            table = self.update_table_deletion_protection_enabled(
+                name, deletion_protection_enabled
+            )
         return table
 
     def update_table_throughput(self, name: str, throughput: Dict[str, int]) -> Table:
@@ -161,6 +171,13 @@ def update_table_billing_mode(self, name: str, billing_mode: str) -> Table:
         table.billing_mode = billing_mode
         return table
 
+    def update_table_deletion_protection_enabled(
+        self, name: str, deletion_protection_enabled: bool
+    ) -> Table:
+        table = self.tables[name]
+        table.deletion_protection_enabled = deletion_protection_enabled
+        return table
+
     def update_table_streams(
         self, name: str, stream_specification: Dict[str, Any]
     ) -> Table:

moto/dynamodb/models/table.py

@@ -243,6 +243,7 @@ def __init__(
         streams: Optional[Dict[str, Any]] = None,
         sse_specification: Optional[Dict[str, Any]] = None,
         tags: Optional[List[Dict[str, str]]] = None,
+        deletion_protection_enabled: Optional[bool] = False,
     ):
         self.name = table_name
         self.account_id = account_id
@@ -306,6 +307,7 @@ def __init__(
             self.sse_specification["KMSMasterKeyId"] = self._get_default_encryption_key(
                 account_id, region
             )
+        self.deletion_protection_enabled = deletion_protection_enabled
 
     def _get_default_encryption_key(self, account_id: str, region: str) -> str:
         from moto.kms import kms_backends
@@ -443,6 +445,7 @@ def describe(self, base_key: str = "TableDescription") -> Dict[str, Any]:
                     index.describe() for index in self.global_indexes
                 ],
                 "LocalSecondaryIndexes": [index.describe() for index in self.indexes],
+                "DeletionProtectionEnabled": self.deletion_protection_enabled,
             }
         }
         if self.latest_stream_label:

moto/dynamodb/responses.py

@@ -302,6 +302,7 @@ def create_table(self) -> str:
         streams = body.get("StreamSpecification")
         # Get any tags
         tags = body.get("Tags", [])
+        deletion_protection_enabled = body.get("DeletionProtectionEnabled", False)
 
         table = self.dynamodb_backend.create_table(
             table_name,
@@ -314,6 +315,7 @@ def create_table(self) -> str:
             billing_mode=billing_mode,
             sse_specification=sse_spec,
             tags=tags,
+            deletion_protection_enabled=deletion_protection_enabled,
         )
         return dynamo_json_dump(table.describe())
 
@@ -431,13 +433,15 @@ def update_table(self) -> str:
         throughput = self.body.get("ProvisionedThroughput", None)
         billing_mode = self.body.get("BillingMode", None)
         stream_spec = self.body.get("StreamSpecification", None)
+        deletion_protection_enabled = self.body.get("DeletionProtectionEnabled")
         table = self.dynamodb_backend.update_table(
             name=name,
             attr_definitions=attr_definitions,
             global_index=global_index,
             throughput=throughput,
             billing_mode=billing_mode,
             stream_spec=stream_spec,
+            deletion_protection_enabled=deletion_protection_enabled,
         )
         return dynamo_json_dump(table.describe())
 

tests/test_dynamodb/exceptions/test_dynamodb_exceptions.py

@@ -1389,3 +1389,31 @@ def test_cannot_scan_gsi_with_consistent_read():
         "Code": "ValidationException",
         "Message": "Consistent reads are not supported on global secondary indexes",
     }
+
+
+@mock_aws
+def test_delete_table():
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    # Create the DynamoDB table.
+    client.create_table(
+        TableName="test1",
+        AttributeDefinitions=[
+            {"AttributeName": "client", "AttributeType": "S"},
+            {"AttributeName": "app", "AttributeType": "S"},
+        ],
+        KeySchema=[
+            {"AttributeName": "client", "KeyType": "HASH"},
+            {"AttributeName": "app", "KeyType": "RANGE"},
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
+        DeletionProtectionEnabled=True,
+    )
+
+    with pytest.raises(ClientError) as err:
+        client.delete_table(TableName="test1")
+    assert err.value.response["Error"]["Code"] == "ValidationException"
+    assert (
+        err.value.response["Error"]["Message"]
+        == "1 validation error detected: Table 'test1' can't be deleted while DeletionProtectionEnabled is set to True"
+    )

tests/test_dynamodb/test_dynamodb_create_table.py

@@ -50,6 +50,7 @@ def test_create_table_standard():
         {"AttributeName": "subject", "KeyType": "RANGE"},
     ]
     assert actual["ItemCount"] == 0
+    assert not actual["DeletionProtectionEnabled"]
 
 
 @mock_aws
@@ -233,6 +234,22 @@ def test_create_table_with_tags():
     assert resp["Tags"] == [{"Key": "tk", "Value": "tv"}]
 
 
+@mock_aws
+def test_create_table_with_deletion_protection_enabled():
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    client.create_table(
+        TableName="test-deletion_protection",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
+        DeletionProtectionEnabled=True,
+    )
+
+    actual = client.describe_table(TableName="test-deletion_protection")["Table"]
+    assert actual["DeletionProtectionEnabled"]
+
+
 @mock_aws
 def test_create_table_pay_per_request():
     client = boto3.client("dynamodb", region_name="us-east-1")

tests/test_dynamodb/test_dynamodb_update_table.py

@@ -53,6 +53,23 @@ def test_update_table_throughput():
     assert table.provisioned_throughput["WriteCapacityUnits"] == 6
 
 
+@mock_aws
+def test_update_table_deletion_protection_enabled():
+    conn = boto3.resource("dynamodb", region_name="us-west-2")
+    table = conn.create_table(
+        TableName="messages",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
+        BillingMode="PAY_PER_REQUEST",
+        DeletionProtectionEnabled=False,
+    )
+    assert not table.deletion_protection_enabled
+
+    table.update(DeletionProtectionEnabled=True)
+
+    assert table.deletion_protection_enabled
+
+
 @mock_aws
 def test_update_table__enable_stream():
     conn = boto3.client("dynamodb", region_name="us-east-1")