Minimal implementation for continuous backup (#7735)

giograno · web-flow · commit 8ef5e66dc615 · 2023-02-27T10:03:24.000-05:00
diff --git a/localstack/services/dynamodb/provider.py b/localstack/services/dynamodb/provider.py
@@ -29,12 +29,15 @@
     BatchWriteItemInput,
     BatchWriteItemOutput,
     BillingMode,
+    ContinuousBackupsDescription,
+    ContinuousBackupsStatus,
     CreateGlobalTableOutput,
     CreateTableInput,
     CreateTableOutput,
     DeleteItemInput,
     DeleteItemOutput,
     DeleteTableOutput,
+    DescribeContinuousBackupsOutput,
     DescribeGlobalTableOutput,
     DescribeKinesisStreamingDestinationOutput,
     DescribeTableOutput,
@@ -55,6 +58,9 @@
     ListTagsOfResourceOutput,
     NextTokenString,
     PartiQLBatchRequest,
+    PointInTimeRecoveryDescription,
+    PointInTimeRecoverySpecification,
+    PointInTimeRecoveryStatus,
     PositiveIntegerObject,
     ProvisionedThroughputExceededException,
     PutItemInput,
@@ -73,6 +79,7 @@
     ScanInput,
     ScanOutput,
     StreamArn,
+    TableDescription,
     TableName,
     TagKeyList,
     TagList,
@@ -81,6 +88,7 @@
     TransactGetItemsOutput,
     TransactWriteItemsInput,
     TransactWriteItemsOutput,
+    UpdateContinuousBackupsOutput,
     UpdateGlobalTableOutput,
     UpdateItemInput,
     UpdateItemOutput,
@@ -109,7 +117,7 @@
 from localstack.utils.aws import arns, aws_stack
 from localstack.utils.aws.arns import extract_account_id_from_arn, extract_region_from_arn
 from localstack.utils.aws.aws_stack import get_valid_regions_for_service
-from localstack.utils.collections import select_attributes
+from localstack.utils.collections import select_attributes, select_from_typed_dict
 from localstack.utils.common import short_uid, to_bytes
 from localstack.utils.json import BytesEncoder, canonical_json
 from localstack.utils.strings import long_uid, to_str
@@ -506,11 +514,13 @@ def describe_table(self, context: RequestContext, table_name: TableName) -> Desc
         global_table_region = self.get_global_table_region(context, table_name)
 
         result = self._forward_request(context=context, region=global_table_region)
+        table_description: TableDescription = result["Table"]
 
         # Update table properties from LocalStack stores
-        table_props = get_store(context.account_id, context.region).table_properties.get(table_name)
-        if table_props:
-            result.get("Table", {}).update(table_props)
+        if table_props := get_store(context.account_id, context.region).table_properties.get(
+            table_name
+        ):
+            table_description.update(table_props)
 
         store = get_store(context.account_id, context.region)
 
@@ -530,22 +540,21 @@ def describe_table(self, context: RequestContext, table_name: TableName) -> Desc
                         RegionName=replicated_region, ReplicaStatus=ReplicaStatus.ACTIVE
                     )
                 )
-        result.get("Table", {}).update({"Replicas": replica_description_list})
+        table_description.update({"Replicas": replica_description_list})
 
         # update only TableId and SSEDescription if present
-        table_definitions = get_store(context.account_id, context.region).table_definitions.get(
-            table_name
-        )
-        if table_definitions:
+        if table_definitions := store.table_definitions.get(table_name):
             for key in ["TableId", "SSEDescription"]:
                 if table_definitions.get(key):
-                    result.get("Table", {})[key] = table_definitions[key]
+                    table_description[key] = table_definitions[key]
             if "TableClass" in table_definitions:
-                result.get("Table", {})["TableClassSummary"] = {
+                table_description["TableClassSummary"] = {
                     "TableClass": table_definitions["TableClass"]
                 }
 
-        return result
+        return DescribeTableOutput(
+            Table=select_from_typed_dict(TableDescription, table_description)
+        )
 
     @handler("UpdateTable", expand=False)
     def update_table(
@@ -1197,6 +1206,55 @@ def describe_kinesis_streaming_destination(
             KinesisDataStreamDestinations=stream_destinations, TableName=table_name
         )
 
+    #
+    # Continuous Backups
+    #
+
+    def describe_continuous_backups(
+        self, context: RequestContext, table_name: TableName
+    ) -> DescribeContinuousBackupsOutput:
+        self.get_global_table_region(context, table_name)
+        store = get_store(context.account_id, context.region)
+        continuous_backup_description = (
+            store.table_properties.get(table_name, {}).get("ContinuousBackupsDescription")
+        ) or ContinuousBackupsDescription(
+            ContinuousBackupsStatus=ContinuousBackupsStatus.ENABLED,
+            PointInTimeRecoveryDescription=PointInTimeRecoveryDescription(
+                PointInTimeRecoveryStatus=PointInTimeRecoveryStatus.DISABLED
+            ),
+        )
+
+        return DescribeContinuousBackupsOutput(
+            ContinuousBackupsDescription=continuous_backup_description
+        )
+
+    def update_continuous_backups(
+        self,
+        context: RequestContext,
+        table_name: TableName,
+        point_in_time_recovery_specification: PointInTimeRecoverySpecification,
+    ) -> UpdateContinuousBackupsOutput:
+        self.get_global_table_region(context, table_name)
+
+        store = get_store(context.account_id, context.region)
+        pit_recovery_status = (
+            PointInTimeRecoveryStatus.ENABLED
+            if point_in_time_recovery_specification["PointInTimeRecoveryEnabled"]
+            else PointInTimeRecoveryStatus.DISABLED
+        )
+        continuous_backup_description = ContinuousBackupsDescription(
+            ContinuousBackupsStatus=ContinuousBackupsStatus.ENABLED,
+            PointInTimeRecoveryDescription=PointInTimeRecoveryDescription(
+                PointInTimeRecoveryStatus=pit_recovery_status
+            ),
+        )
+        table_props = store.table_properties.setdefault(table_name, {})
+        table_props["ContinuousBackupsDescription"] = continuous_backup_description
+
+        return UpdateContinuousBackupsOutput(
+            ContinuousBackupsDescription=continuous_backup_description
+        )
+
     #
     # Helpers
     #
@@ -1271,7 +1329,7 @@ def prepare_request_headers(headers: Dict, account_id: str, region_name: str):
 
         # DynamoDBLocal namespaces based on the value of Credentials
         # Since we want to namespace by both account ID and region, use an aggregate key
-        # We also replace the region to keep compatibilty with NoSQL Workbench
+        # We also replace the region to keep compatibility with NoSQL Workbench
         headers["Authorization"] = re.sub(
             AUTH_CREDENTIAL_REGEX,
             rf"Credential={key}/\2/{region_name}/\4/",
diff --git a/tests/integration/test_dynamodb.py b/tests/integration/test_dynamodb.py
@@ -8,12 +8,17 @@
 from boto3.dynamodb.conditions import Key
 from boto3.dynamodb.types import STRING
 
+from localstack.aws.api.dynamodb import (
+    ContinuousBackupsUnavailableException,
+    PointInTimeRecoverySpecification,
+)
 from localstack.constants import TEST_AWS_SECRET_ACCESS_KEY
 from localstack.services.dynamodbstreams.dynamodbstreams_api import get_kinesis_stream_name
 from localstack.testing.snapshots.transformer import SortingTransformer
 from localstack.utils import testutil
 from localstack.utils.aws import arns, aws_stack, queries, resources
 from localstack.utils.common import json_safe, long_uid, retry, short_uid
+from localstack.utils.sync import poll_condition
 
 from .test_kinesis import get_shard_iterator
 
@@ -1610,6 +1615,45 @@ def test_data_encoding_consistency(
         ]
         snapshot.match("GetRecordsAfterUpdate", records[1]["dynamodb"]["NewImage"])
 
+    @pytest.mark.skip_snapshot_verify(
+        paths=[
+            "$..PointInTimeRecoveryDescription..EarliestRestorableDateTime",
+            "$..PointInTimeRecoveryDescription..LatestRestorableDateTime",
+        ]
+    )
+    def test_continuous_backup_update(self, dynamodb_create_table, dynamodb_client, snapshot):
+        table_name = f"table-{short_uid()}"
+        dynamodb_create_table(
+            table_name=table_name,
+            partition_key=PARTITION_KEY,
+        )
+
+        def wait_for_continuous_backend():
+            try:
+                dynamodb_client.update_continuous_backups(
+                    TableName=table_name,
+                    PointInTimeRecoverySpecification=PointInTimeRecoverySpecification(
+                        PointInTimeRecoveryEnabled=True
+                    ),
+                )
+                return True
+            except ContinuousBackupsUnavailableException:
+                return False
+
+        assert poll_condition(wait_for_continuous_backend, timeout=10)
+
+        response = dynamodb_client.update_continuous_backups(
+            TableName=table_name,
+            PointInTimeRecoverySpecification=PointInTimeRecoverySpecification(
+                PointInTimeRecoveryEnabled=True
+            ),
+        )
+
+        snapshot.match("update-continuous-backup", response)
+
+        response = dynamodb_client.describe_continuous_backups(TableName=table_name)
+        snapshot.match("describe-continuous-backup", response)
+
 
 def delete_table(name):
     dynamodb_client = aws_stack.create_external_boto_client("dynamodb")
diff --git a/tests/integration/test_dynamodb.snapshot.json b/tests/integration/test_dynamodb.snapshot.json
@@ -376,5 +376,38 @@
         }
       }
     }
+  },
+  "tests/integration/test_dynamodb.py::TestDynamoDB::test_continuous_backup_update": {
+    "recorded-date": "22-02-2023, 21:13:32",
+    "recorded-content": {
+      "update-continuous-backup": {
+        "ContinuousBackupsDescription": {
+          "ContinuousBackupsStatus": "ENABLED",
+          "PointInTimeRecoveryDescription": {
+            "EarliestRestorableDateTime": "datetime",
+            "LatestRestorableDateTime": "datetime",
+            "PointInTimeRecoveryStatus": "ENABLED"
+          }
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      },
+      "describe-continuous-backup": {
+        "ContinuousBackupsDescription": {
+          "ContinuousBackupsStatus": "ENABLED",
+          "PointInTimeRecoveryDescription": {
+            "EarliestRestorableDateTime": "datetime",
+            "LatestRestorableDateTime": "datetime",
+            "PointInTimeRecoveryStatus": "ENABLED"
+          }
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
   }
-}
+}
