DynamoDB: CloudFormation attributes for creating Table (#7674)

moto/dynamodb/models/__init__.py

@@ -60,11 +60,36 @@ def default_vpc_endpoint_service(
             base_endpoint_dns_names=[f"dynamodb.{service_region}.amazonaws.com"],
         )
 
-    def create_table(self, name: str, **params: Any) -> Table:
+    def create_table(
+        self,
+        name: str,
+        schema: List[Dict[str, str]],
+        throughput: Optional[Dict[str, int]],
+        attr: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        indexes: Optional[List[Dict[str, Any]]],
+        streams: Optional[Dict[str, Any]],
+        billing_mode: str,
+        sse_specification: Optional[Dict[str, Any]],
+        tags: List[Dict[str, str]],
+        deletion_protection_enabled: bool,
+    ) -> Table:
         if name in self.tables:
             raise ResourceInUseException(f"Table already exists: {name}")
         table = Table(
-            name, account_id=self.account_id, region=self.region_name, **params
+            name,
+            account_id=self.account_id,
+            region=self.region_name,
+            schema=schema,
+            throughput=throughput,
+            attr=attr,
+            global_indexes=global_indexes,
+            indexes=indexes,
+            streams=streams,
+            billing_mode=billing_mode,
+            sse_specification=sse_specification,
+            tags=tags,
+            deletion_protection_enabled=deletion_protection_enabled,
         )
         self.tables[name] = table
         return table

moto/dynamodb/models/table.py

@@ -386,14 +386,22 @@ def create_from_cloudformation_json(  # type: ignore[misc]
             params["schema"] = properties["KeySchema"]
         if "AttributeDefinitions" in properties:
             params["attr"] = properties["AttributeDefinitions"]
-        if "GlobalSecondaryIndexes" in properties:
-            params["global_indexes"] = properties["GlobalSecondaryIndexes"]
-        if "ProvisionedThroughput" in properties:
-            params["throughput"] = properties["ProvisionedThroughput"]
-        if "LocalSecondaryIndexes" in properties:
-            params["indexes"] = properties["LocalSecondaryIndexes"]
-        if "StreamSpecification" in properties:
-            params["streams"] = properties["StreamSpecification"]
+        params["global_indexes"] = properties.get("GlobalSecondaryIndexes", [])
+        params["throughput"] = properties.get("ProvisionedThroughput")
+        params["indexes"] = properties.get("LocalSecondaryIndexes", [])
+        params["streams"] = properties.get("StreamSpecification")
+        params["tags"] = properties.get("Tags")
+        params["deletion_protection_enabled"] = properties.get(
+            "DeletionProtectionEnabled", False
+        )
+        params["sse_specification"] = properties.get("SSESpecification")
+
+        billing_mode = (
+            "PAY_PER_REQUEST"
+            if properties.get("BillingMode") == "PAY_PER_REQUEST"
+            else "PROVISIONED"
+        )
+        params["billing_mode"] = billing_mode
 
         table = dynamodb_backends[account_id][region_name].create_table(
             name=resource_name, **params

tests/test_dynamodb/test_dynamodb_cloudformation.py

@@ -1,4 +1,5 @@
 import json
+from copy import deepcopy
 
 import boto3
 
@@ -7,7 +8,7 @@
 template_create_table = {
     "AWSTemplateFormatVersion": "2010-09-09",
     "Resources": {
-        "myDynamoDBTable": {
+        "table": {
             "Type": "AWS::DynamoDB::Table",
             "Properties": {
                 "AttributeDefinitions": [
@@ -30,7 +31,56 @@
 
 
 @mock_aws
-def test_delete_stack_dynamo_template_boto3():
+def test_create_stack_pay_per_request():
+    conn = boto3.client("cloudformation", region_name="us-east-1")
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    template = deepcopy(template_create_table)
+    template["Resources"]["table"]["Properties"]["BillingMode"] = "PAY_PER_REQUEST"
+    del template["Resources"]["table"]["Properties"]["ProvisionedThroughput"]
+
+    conn.create_stack(StackName="test", TemplateBody=json.dumps(template))
+    table_desc = dynamodb_client.list_tables()
+    assert len(table_desc.get("TableNames")) == 1
+
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert table["BillingModeSummary"] == {"BillingMode": "PAY_PER_REQUEST"}
+
+
+@mock_aws
+def test_create_stack_with_indexes():
+    conn = boto3.client("cloudformation", region_name="us-east-1")
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    template = deepcopy(template_create_table)
+    template["Resources"]["table"]["Properties"]["GlobalSecondaryIndexes"] = [
+        {
+            "IndexName": "gsi",
+            "KeySchema": [{"AttributeName": "gsipk", "KeyType": "S"}],
+            "Projection": {"ProjectionType": "ALL"},
+        }
+    ]
+    template["Resources"]["table"]["Properties"]["LocalSecondaryIndexes"] = [
+        {
+            "IndexName": "lsi",
+            "KeySchema": [{"AttributeName": "lsipk", "KeyType": "S"}],
+            "Projection": {"ProjectionType": "ALL"},
+        }
+    ]
+
+    conn.create_stack(StackName="test", TemplateBody=json.dumps(template))
+    table_desc = dynamodb_client.list_tables()
+    assert len(table_desc.get("TableNames")) == 1
+
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert len(table["GlobalSecondaryIndexes"]) == 1
+    assert len(table["LocalSecondaryIndexes"]) == 1
+
+
+@mock_aws
+def test_delete_stack_dynamo_template():
     conn = boto3.client("cloudformation", region_name="us-east-1")
     dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
 
@@ -40,6 +90,19 @@ def test_delete_stack_dynamo_template_boto3():
     table_desc = dynamodb_client.list_tables()
     assert len(table_desc.get("TableNames")) == 1
 
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert table["ProvisionedThroughput"] == {
+        "NumberOfDecreasesToday": 0,
+        "ReadCapacityUnits": 5,
+        "WriteCapacityUnits": 5,
+    }
+    assert table["BillingModeSummary"] == {"BillingMode": "PROVISIONED"}
+    assert table["LocalSecondaryIndexes"] == []
+    assert table["GlobalSecondaryIndexes"] == []
+    assert table["DeletionProtectionEnabled"] is False
+
     conn.delete_stack(StackName="test_stack")
     table_desc = dynamodb_client.list_tables()
     assert len(table_desc.get("TableNames")) == 0