text: from __future__ import unicode_literals
  
  import datetime
  
  from botocore.exceptions import ClientError
  import boto3
  import sure  # noqa
  
  from moto import mock_kinesis
  from moto.core import ACCOUNT_ID
  
  
  def create_s3_delivery_stream(client, stream_name):
      return client.create_delivery_stream(
          DeliveryStreamName=stream_name,
          DeliveryStreamType="DirectPut",
          ExtendedS3DestinationConfiguration={
              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(ACCOUNT_ID),
              "BucketARN": "arn:aws:s3:::kinesis-test",
              "Prefix": "myFolder/",
              "CompressionFormat": "UNCOMPRESSED",
              "DataFormatConversionConfiguration": {
                  "Enabled": True,
                  "InputFormatConfiguration": {"Deserializer": {"HiveJsonSerDe": {}}},
                  "OutputFormatConfiguration": {
                      "Serializer": {"ParquetSerDe": {"Compression": "SNAPPY"}}
                  },
                  "SchemaConfiguration": {
                      "DatabaseName": stream_name,
                      "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                          ACCOUNT_ID
                      ),
                      "TableName": "outputTable",
                  },
              },
          },
      )
  
  
  def create_redshift_delivery_stream(client, stream_name):
      return client.create_delivery_stream(
          DeliveryStreamName=stream_name,
          RedshiftDestinationConfiguration={
              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(ACCOUNT_ID),
              "ClusterJDBCURL": "jdbc:redshift://host.amazonaws.com:5439/database",
              "CopyCommand": {
                  "DataTableName": "outputTable",
                  "CopyOptions": "CSV DELIMITER ',' NULL '\\0'",
              },
              "Username": "username",
              "Password": "password",
              "S3Configuration": {
                  "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                      ACCOUNT_ID
                  ),
                  "BucketARN": "arn:aws:s3:::kinesis-test",
                  "Prefix": "myFolder/",
                  "BufferingHints": {"SizeInMBs": 123, "IntervalInSeconds": 124},
                  "CompressionFormat": "UNCOMPRESSED",
              },
          },
      )
  
  
  def create_elasticsearch_delivery_stream(client, stream_name):
      return client.create_delivery_stream(
          DeliveryStreamName=stream_name,
          DeliveryStreamType="DirectPut",
          ElasticsearchDestinationConfiguration={
              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(ACCOUNT_ID),
              "DomainARN": "arn:aws:es:::domain/kinesis-test",
              "IndexName": "myIndex",
              "TypeName": "UNCOMPRESSED",
              "IndexRotationPeriod": "NoRotation",
              "BufferingHints": {"IntervalInSeconds": 123, "SizeInMBs": 123},
              "RetryOptions": {"DurationInSeconds": 123},
              "S3Configuration": {
                  "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                      ACCOUNT_ID
                  ),
                  "BucketARN": "arn:aws:s3:::kinesis-test",
                  "Prefix": "myFolder/",
                  "BufferingHints": {"SizeInMBs": 123, "IntervalInSeconds": 124},
                  "CompressionFormat": "UNCOMPRESSED",
              },
          },
      )
  
  
  @mock_kinesis
  def test_create_redshift_delivery_stream():
      client = boto3.client("firehose", region_name="us-east-1")
  
      response = create_redshift_delivery_stream(client, "stream1")
      stream_arn = response["DeliveryStreamARN"]
  
      response = client.describe_delivery_stream(DeliveryStreamName="stream1")
      stream_description = response["DeliveryStreamDescription"]
  
      # Sure and Freezegun don't play nicely together
      _ = stream_description.pop("CreateTimestamp")
      _ = stream_description.pop("LastUpdateTimestamp")
  
      stream_description.should.equal(
          {
              "DeliveryStreamName": "stream1",
              "DeliveryStreamARN": stream_arn,
              "DeliveryStreamStatus": "ACTIVE",
              "VersionId": "string",
              "Destinations": [
                  {
                      "DestinationId": "string",
                      "RedshiftDestinationDescription": {
                          "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                              ACCOUNT_ID
                          ),
                          "ClusterJDBCURL": "jdbc:redshift://host.amazonaws.com:5439/database",
                          "CopyCommand": {
                              "DataTableName": "outputTable",
                              "CopyOptions": "CSV DELIMITER ',' NULL '\\0'",
                          },
                          "Username": "username",
                          "S3DestinationDescription": {
                              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                                  ACCOUNT_ID
                              ),
                              "BucketARN": "arn:aws:s3:::kinesis-test",
                              "Prefix": "myFolder/",
                              "BufferingHints": {
                                  "SizeInMBs": 123,
                                  "IntervalInSeconds": 124,
                              },
                              "CompressionFormat": "UNCOMPRESSED",
                          },
                      },
                  }
              ],
              "HasMoreDestinations": False,
          }
      )
  
  
  @mock_kinesis
  def test_create_s3_delivery_stream():
      client = boto3.client("firehose", region_name="us-east-1")
  
      response = create_s3_delivery_stream(client, "stream1")
      stream_arn = response["DeliveryStreamARN"]
  
      response = client.describe_delivery_stream(DeliveryStreamName="stream1")
      stream_description = response["DeliveryStreamDescription"]
  
      # Sure and Freezegun don't play nicely together
      _ = stream_description.pop("CreateTimestamp")
      _ = stream_description.pop("LastUpdateTimestamp")
  
      stream_description.should.equal(
          {
              "DeliveryStreamName": "stream1",
              "DeliveryStreamARN": stream_arn,
              "DeliveryStreamStatus": "ACTIVE",
              "VersionId": "string",
              "Destinations": [
                  {
                      "DestinationId": "string",
                      "ExtendedS3DestinationDescription": {
                          "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                              ACCOUNT_ID
                          ),
                          "BucketARN": "arn:aws:s3:::kinesis-test",
                          "Prefix": "myFolder/",
                          "CompressionFormat": "UNCOMPRESSED",
                          "DataFormatConversionConfiguration": {
                              "Enabled": True,
                              "InputFormatConfiguration": {
                                  "Deserializer": {"HiveJsonSerDe": {}}
                              },
                              "OutputFormatConfiguration": {
                                  "Serializer": {
                                      "ParquetSerDe": {"Compression": "SNAPPY"}
                                  }
                              },
                              "SchemaConfiguration": {
                                  "DatabaseName": "stream1",
                                  "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                                      ACCOUNT_ID
                                  ),
                                  "TableName": "outputTable",
                              },
                          },
                      },
                  }
              ],
              "HasMoreDestinations": False,
          }
      )
  
  
  @mock_kinesis
  def test_create_elasticsearch_delivery_stream():
      client = boto3.client("firehose", region_name="us-east-1")
  
      response = create_elasticsearch_delivery_stream(client, "stream1")
      stream_arn = response["DeliveryStreamARN"]
  
      response = client.describe_delivery_stream(DeliveryStreamName="stream1")
      stream_description = response["DeliveryStreamDescription"]
  
      # Sure and Freezegun don't play nicely together
      _ = stream_description.pop("CreateTimestamp")
      _ = stream_description.pop("LastUpdateTimestamp")
  
      stream_description.should.equal(
          {
              "DeliveryStreamName": "stream1",
              "DeliveryStreamARN": stream_arn,
              "DeliveryStreamStatus": "ACTIVE",
              "VersionId": "string",
              "Destinations": [
                  {
                      "DestinationId": "string",
                      "ElasticsearchDestinationDescription": {
                          "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                              ACCOUNT_ID
                          ),
                          "DomainARN": "arn:aws:es:::domain/kinesis-test",
                          "IndexName": "myIndex",
                          "TypeName": "UNCOMPRESSED",
                          "IndexRotationPeriod": "NoRotation",
                          "BufferingHints": {"IntervalInSeconds": 123, "SizeInMBs": 123},
                          "RetryOptions": {"DurationInSeconds": 123},
                          "S3DestinationDescription": {
                              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                                  ACCOUNT_ID
                              ),
                              "BucketARN": "arn:aws:s3:::kinesis-test",
                              "Prefix": "myFolder/",
                              "BufferingHints": {
                                  "SizeInMBs": 123,
                                  "IntervalInSeconds": 124,
                              },
                              "CompressionFormat": "UNCOMPRESSED",
                          },
                      },
                  }
              ],
              "HasMoreDestinations": False,
          }
      )
  
  
  @mock_kinesis
  def test_create_stream_without_redshift():
      client = boto3.client("firehose", region_name="us-east-1")
  
      response = client.create_delivery_stream(
          DeliveryStreamName="stream1",
          S3DestinationConfiguration={
              "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(ACCOUNT_ID),
              "BucketARN": "arn:aws:s3:::kinesis-test",
              "Prefix": "myFolder/",
              "BufferingHints": {"SizeInMBs": 123, "IntervalInSeconds": 124},
              "CompressionFormat": "UNCOMPRESSED",
          },
      )
      stream_arn = response["DeliveryStreamARN"]
  
      response = client.describe_delivery_stream(DeliveryStreamName="stream1")
      stream_description = response["DeliveryStreamDescription"]
  
      # Sure and Freezegun don't play nicely together
      _ = stream_description.pop("CreateTimestamp")
      _ = stream_description.pop("LastUpdateTimestamp")
  
      stream_description.should.equal(
          {
              "DeliveryStreamName": "stream1",
              "DeliveryStreamARN": stream_arn,
              "DeliveryStreamStatus": "ACTIVE",
              "VersionId": "string",
              "Destinations": [
                  {
                      "DestinationId": "string",
                      "S3DestinationDescription": {
                          "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                              ACCOUNT_ID
                          ),
                          "RoleARN": "arn:aws:iam::{}:role/firehose_delivery_role".format(
                              ACCOUNT_ID
                          ),
                          "BucketARN": "arn:aws:s3:::kinesis-test",
                          "Prefix": "myFolder/",
                          "BufferingHints": {"SizeInMBs": 123, "IntervalInSeconds": 124},
                          "CompressionFormat": "UNCOMPRESSED",
                      },
                  }
              ],
              "HasMoreDestinations": False,
          }
      )
  
  
  @mock_kinesis
  def test_deescribe_non_existent_stream():
      client = boto3.client("firehose", region_name="us-east-1")
  
      client.describe_delivery_stream.when.called_with(
          DeliveryStreamName="not-a-stream"
      ).should.throw(ClientError)
  
  
  @mock_kinesis
  def test_list_and_delete_stream():
      client = boto3.client("firehose", region_name="us-east-1")
  
      create_redshift_delivery_stream(client, "stream1")
      create_redshift_delivery_stream(client, "stream2")
  
      set(client.list_delivery_streams()["DeliveryStreamNames"]).should.equal(
          set(["stream1", "stream2"])
      )
  
      client.delete_delivery_stream(DeliveryStreamName="stream1")
  
      set(client.list_delivery_streams()["DeliveryStreamNames"]).should.equal(
          set(["stream2"])
      )
  
  
  @mock_kinesis
  def test_put_record():
      client = boto3.client("firehose", region_name="us-east-1")
  
      create_redshift_delivery_stream(client, "stream1")
      client.put_record(DeliveryStreamName="stream1", Record={"Data": "some data"})
  
  
  @mock_kinesis
  def test_put_record_batch():
      client = boto3.client("firehose", region_name="us-east-1")
  
      create_redshift_delivery_stream(client, "stream1")
      client.put_record_batch(
          DeliveryStreamName="stream1",
          Records=[{"Data": "some data1"}, {"Data": "some data2"}],
      )
  