Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use assert_s3_contents method in localstack

Best Python code snippet using localstack_python

test_firehose.py

Source:test_firehose.py

...206                sources = [item["_source"] for item in result]207                assert firehose_record in sources208                assert kinesis_record in sources209            retry(assert_elasticsearch_contents)210            def assert_s3_contents():211                result = s3_client.list_objects(Bucket=s3_bucket)212                contents = []213                for o in result.get("Contents"):214                    data = s3_client.get_object(Bucket=s3_bucket, Key=o.get("Key"))215                    content = data["Body"].read()216                    contents.append(content)217                assert len(contents) == 2218                assert to_bytes(json.dumps(firehose_record)) in contents219                assert to_bytes(json.dumps(kinesis_record)) in contents220            retry(assert_s3_contents)221        finally:222            firehose_client.delete_delivery_stream(DeliveryStreamName=delivery_stream_name)223            es_client.delete_elasticsearch_domain(DomainName=domain_name)224    @pytest.mark.skip_offline225    @pytest.mark.parametrize("opensearch_endpoint_strategy", ["domain", "path"])226    def test_kinesis_firehose_opensearch_s3_backup(227        self,228        firehose_client,229        kinesis_client,230        opensearch_client,231        s3_client,232        s3_bucket,233        kinesis_create_stream,234        monkeypatch,235        opensearch_endpoint_strategy,236    ):237        domain_name = f"test-domain-{short_uid()}"238        stream_name = f"test-stream-{short_uid()}"239        role_arn = "arn:aws:iam::000000000000:role/Firehose-Role"240        delivery_stream_name = f"test-delivery-stream-{short_uid()}"241        monkeypatch.setattr(config, "OPENSEARCH_ENDPOINT_STRATEGY", opensearch_endpoint_strategy)242        try:243            opensearch_create_response = opensearch_client.create_domain(DomainName=domain_name)244            opensearch_url = f"http://{opensearch_create_response['DomainStatus']['Endpoint']}"245            opensearch_arn = opensearch_create_response["DomainStatus"]["ARN"]246            # create s3 backup bucket arn247            bucket_arn = aws_stack.s3_bucket_arn(s3_bucket)248            # create kinesis stream249            kinesis_create_stream(StreamName=stream_name, ShardCount=2)250            stream_arn = kinesis_client.describe_stream(StreamName=stream_name)[251                "StreamDescription"252            ]["StreamARN"]253            kinesis_stream_source_def = {254                "KinesisStreamARN": stream_arn,255                "RoleARN": role_arn,256            }257            opensearch_destination_configuration = {258                "RoleARN": role_arn,259                "DomainARN": opensearch_arn,260                "IndexName": "activity",261                "TypeName": "activity",262                "S3BackupMode": "AllDocuments",263                "S3Configuration": {264                    "RoleARN": role_arn,265                    "BucketARN": bucket_arn,266                },267            }268            firehose_client.create_delivery_stream(269                DeliveryStreamName=delivery_stream_name,270                DeliveryStreamType="KinesisStreamAsSource",271                KinesisStreamSourceConfiguration=kinesis_stream_source_def,272                AmazonopensearchserviceDestinationConfiguration=opensearch_destination_configuration,273            )274            # wait for opensearch cluster to be ready275            def check_domain_state():276                result = opensearch_client.describe_domain(DomainName=domain_name)["DomainStatus"][277                    "Processing"278                ]279                return not result280            assert poll_condition(check_domain_state, 30, 1)281            # put kinesis stream record282            kinesis_record = {"target": "hello"}283            kinesis_client.put_record(284                StreamName=stream_name, Data=to_bytes(json.dumps(kinesis_record)), PartitionKey="1"285            )286            firehose_record = {"target": "world"}287            firehose_client.put_record(288                DeliveryStreamName=delivery_stream_name,289                Record={"Data": to_bytes(json.dumps(firehose_record))},290            )291            def assert_opensearch_contents():292                response = requests.get(f"{opensearch_url}/activity/_search")293                response_bod = response.json()294                assert "hits" in response_bod295                response_bod_hits = response_bod["hits"]296                assert "hits" in response_bod_hits297                result = response_bod_hits["hits"]298                assert len(result) == 2299                sources = [item["_source"] for item in result]300                assert firehose_record in sources301                assert kinesis_record in sources302            retry(assert_opensearch_contents)303            def assert_s3_contents():304                result = s3_client.list_objects(Bucket=s3_bucket)305                contents = []306                for o in result.get("Contents"):307                    data = s3_client.get_object(Bucket=s3_bucket, Key=o.get("Key"))308                    content = data["Body"].read()309                    contents.append(content)310                assert len(contents) == 2311                assert to_bytes(json.dumps(firehose_record)) in contents312                assert to_bytes(json.dumps(kinesis_record)) in contents313            retry(assert_s3_contents)314        finally:315            firehose_client.delete_delivery_stream(DeliveryStreamName=delivery_stream_name)316            opensearch_client.delete_domain(DomainName=domain_name)317    def test_delivery_stream_with_kinesis_as_source(...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.