Firehose: Add Snowflake Destination (#8122)

moto/firehose/models.py

@@ -47,6 +47,7 @@
     "http_endpoint": "HttpEndpoint",
     "elasticsearch": "Elasticsearch",
     "redshift": "Redshift",
+    "snowflake": "Snowflake",
     "splunk": "Splunk",  # Unimplemented
 }
 
@@ -200,6 +201,7 @@ def create_delivery_stream(  # pylint: disable=unused-argument
         elasticsearch_destination_configuration: Dict[str, Any],
         splunk_destination_configuration: Dict[str, Any],
         http_endpoint_destination_configuration: Dict[str, Any],
+        snowflake_destination_configuration: Dict[str, Any],
         tags: List[Dict[str, str]],
     ) -> str:
         """Create a Kinesis Data Firehose delivery stream."""
@@ -600,6 +602,7 @@ def update_destination(  # pylint: disable=unused-argument
         elasticsearch_destination_update: Dict[str, Any],
         splunk_destination_update: Dict[str, Any],
         http_endpoint_destination_update: Dict[str, Any],
+        snowflake_destination_configuration: Dict[str, Any],
     ) -> None:
         (dest_name, dest_config) = find_destination_config_in_args(locals())
 

moto/firehose/responses.py

@@ -32,6 +32,7 @@ def create_delivery_stream(self) -> str:
             self._get_param("ElasticsearchDestinationConfiguration"),
             self._get_param("SplunkDestinationConfiguration"),
             self._get_param("HttpEndpointDestinationConfiguration"),
+            self._get_param("SnowflakeDestinationConfiguration"),
             self._get_param("Tags"),
         )
         return json.dumps({"DeliveryStreamARN": delivery_stream_arn})
@@ -109,6 +110,7 @@ def update_destination(self) -> str:
             self._get_param("ElasticsearchDestinationUpdate"),
             self._get_param("SplunkDestinationUpdate"),
             self._get_param("HttpEndpointDestinationUpdate"),
+            self._get_param("SnowflakeDestinationConfiguration"),
         )
         return json.dumps({})
 

tests/test_firehose/test_firehose_destination_types.py

@@ -102,6 +102,69 @@ def create_http_delivery_stream(client, stream_name):
     )
 
 
+def create_snowflake_delivery_stream(client, stream_name):
+    """Return delivery stream ARN of a Snowflake destination."""
+    return client.create_delivery_stream(
+        DeliveryStreamName=stream_name,
+        DeliveryStreamType="DirectPut",
+        SnowflakeDestinationConfiguration={
+            "RoleARN": f"arn:aws:iam::{ACCOUNT_ID}:role/firehose_delivery_role",
+            "AccountUrl": f"fake-account.{TEST_REGION}.snowflakecomputing.com",
+            "Database": "myDatabase",
+            "Schema": "mySchema",
+            "Table": "myTable",
+            "S3BackupMode": "FailedDataOnly",
+            "S3Configuration": {
+                "RoleARN": f"arn:aws:iam::{ACCOUNT_ID}:role/firehose_delivery_role",
+                "BucketARN": "arn:aws:s3:::firehose-test",
+            },
+        },
+    )
+
+
+@mock_aws
+def test_create_snowflake_delivery_stream():
+    """Verify fields of a Snowflake delivery stream."""
+    client = boto3.client("firehose", region_name=TEST_REGION)
+
+    stream_name = f"stream_{mock_random.get_random_hex(6)}"
+    response = create_snowflake_delivery_stream(client, stream_name)
+    stream_arn = response["DeliveryStreamARN"]
+
+    response = client.describe_delivery_stream(DeliveryStreamName=stream_name)
+    stream_description = response["DeliveryStreamDescription"]
+
+    # Sure and Freezegun don't play nicely together
+    stream_description.pop("CreateTimestamp")
+    stream_description.pop("LastUpdateTimestamp")
+
+    assert stream_description == {
+        "DeliveryStreamName": stream_name,
+        "DeliveryStreamARN": stream_arn,
+        "DeliveryStreamStatus": "ACTIVE",
+        "DeliveryStreamType": "DirectPut",
+        "Destinations": [
+            {
+                "DestinationId": "destinationId-000000000001",
+                "SnowflakeDestinationDescription": {
+                    "AccountUrl": f"fake-account.{TEST_REGION}.snowflakecomputing.com",
+                    "Database": "myDatabase",
+                    "RoleARN": f"arn:aws:iam::{ACCOUNT_ID}:role/firehose_delivery_role",
+                    "S3BackupMode": "FailedDataOnly",
+                    "S3DestinationDescription": {
+                        "RoleARN": f"arn:aws:iam::{ACCOUNT_ID}:role/firehose_delivery_role",
+                        "BucketARN": "arn:aws:s3:::firehose-test",
+                    },
+                    "Schema": "mySchema",
+                    "Table": "myTable",
+                },
+            },
+        ],
+        "HasMoreDestinations": False,
+        "VersionId": "1",
+    }
+
+
 @mock_aws
 def test_create_redshift_delivery_stream():
     """Verify fields of a Redshift delivery stream."""