moto/tests/test_firehose/test_http_destinations.py

import boto3

from moto import mock_aws
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
from moto.moto_api._internal import mock_random

from .test_firehose_destination_types import create_http_delivery_stream

TEST_REGION = "us-west-1"


@mock_aws
def test_create_http_stream():
    """Verify fields of a HTTP delivery stream."""
    client = boto3.client("firehose", region_name=TEST_REGION)

    stream_name = f"stream_{mock_random.get_random_hex(6)}"
    response = create_http_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",
        "VersionId": "1",
        "Destinations": [
            {
                "DestinationId": "destinationId-000000000001",
                "HttpEndpointDestinationDescription": {
                    "EndpointConfiguration": {"Url": "google.com"},
                    "RetryOptions": {"DurationInSeconds": 100},
                    "BufferingHints": {"SizeInMBs": 123, "IntervalInSeconds": 124},
                    "CloudWatchLoggingOptions": {"Enabled": False},
                    "S3DestinationDescription": {
                        "RoleARN": f"arn:aws:iam::{ACCOUNT_ID}:role/firehose_delivery_role",
                        "BucketARN": "arn:aws:s3:::firehose-test",
                        "Prefix": "myFolder/",
                        "BufferingHints": {
                            "SizeInMBs": 123,
                            "IntervalInSeconds": 124,
                        },
                        "CompressionFormat": "UNCOMPRESSED",
                    },
                },
            }
        ],
        "HasMoreDestinations": False,
    }


@mock_aws
def test_update_s3_for_http_stream():
    """Verify fields of a HTTP delivery stream."""
    client = boto3.client("firehose", region_name=TEST_REGION)

    stream_name = f"stream_{mock_random.get_random_hex(6)}"
    create_http_delivery_stream(client, stream_name)

    stream = client.describe_delivery_stream(DeliveryStreamName=stream_name)
    version_id = stream["DeliveryStreamDescription"]["VersionId"]

    client.update_destination(
        DeliveryStreamName=stream_name,
        CurrentDeliveryStreamVersionId=version_id,
        DestinationId="destinationId-000000000001",
        HttpEndpointDestinationUpdate={"S3Update": {"ErrorOutputPrefix": "prefix2"}},
    )

    desc = client.describe_delivery_stream(DeliveryStreamName=stream_name)[
        "DeliveryStreamDescription"
    ]
    s3_desc = desc["Destinations"][0]["HttpEndpointDestinationDescription"][
        "S3DestinationDescription"
    ]
    assert s3_desc["ErrorOutputPrefix"] == "prefix2"