Firehose - Snowflake destination: Fix put_record and `put_record_ba…

…tch` functions (getmoto#8152)

moto/firehose/models.py

@@ -517,8 +517,8 @@ def put_record_batch(
                 request_responses = self.put_http_records(
                     destination["HttpEndpoint"], records
                 )
-            elif "Elasticsearch" in destination or "Redshift" in destination:
-                # This isn't implmented as these services aren't implemented,
+            elif {"Elasticsearch", "Redshift", "Snowflake"} & set(destination):
+                # This isn't implemented as these services aren't implemented,
                 # so ignore the data, but return a "proper" response.
                 request_responses = [
                     {"RecordId": str(mock_random.uuid4())} for _ in range(len(records))

tests/test_firehose/test_firehose_put.py

@@ -149,3 +149,69 @@ def test_put_record_batch_extended_s3_destination():
     assert re.match(
         r"^[0-9]{4}/[0-9]{2}/[0-9]{2}/[0-9]{2}/", bucket_objects["Contents"][0]["Key"]
     )
+
+
+@mock_aws
+def test_put_record_snowflake_destination():
+    """Test invocations of put_record() to a Snowflake destination."""
+    client = boto3.client("firehose", region_name=TEST_REGION)
+
+    stream_name = f"test_put_record_{mock_random.get_random_hex(6)}"
+    client.create_delivery_stream(
+        DeliveryStreamName=stream_name,
+        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",
+            },
+        },
+    )
+    result = client.put_record(
+        DeliveryStreamName=stream_name, Record={"Data": "some test data"}
+    )
+
+    assert set(result.keys()) == {"RecordId", "Encrypted", "ResponseMetadata"}
+    assert result["Encrypted"] is False
+    assert result["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+
+@mock_aws
+def test_put_record_batch_snowflake_destination():
+    """Test invocations of put_record_batch() to a Snowflake destination."""
+    client = boto3.client("firehose", region_name=TEST_REGION)
+
+    stream_name = f"test_put_record_{mock_random.get_random_hex(6)}"
+    client.create_delivery_stream(
+        DeliveryStreamName=stream_name,
+        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",
+            },
+        },
+    )
+
+    records = [{"Data": "one"}, {"Data": "two"}, {"Data": "three"}]
+    result = client.put_record_batch(DeliveryStreamName=stream_name, Records=records)
+
+    assert set(result.keys()) == {
+        "RequestResponses",
+        "FailedPutCount",
+        "ResponseMetadata",
+        "Encrypted",
+    }
+    assert len(result["RequestResponses"]) == 3
+    assert result["Encrypted"] is False
+    assert result["ResponseMetadata"]["HTTPStatusCode"] == 200