How to use dynamodbstreams_client method in localstack

Best Python code snippet using localstack_python

test_dynamodb.py

Source:test_dynamodb.py

1# -*- coding: utf-8 -*-2import json3import re4from datetime import datetime5from time import sleep6from typing import Dict7import pytest8from boto3.dynamodb.conditions import Key9from boto3.dynamodb.types import STRING10from localstack.services.awslambda.lambda_utils import LAMBDA_RUNTIME_PYTHON3611from localstack.services.dynamodbstreams.dynamodbstreams_api import get_kinesis_stream_name12from localstack.utils import testutil13from localstack.utils.aws import aws_stack14from localstack.utils.common import json_safe, long_uid, retry, short_uid15from localstack.utils.testutil import check_expected_lambda_log_events_length16from .awslambda.test_lambda import TEST_LAMBDA_PYTHON_ECHO17from .test_kinesis import get_shard_iterator18PARTITION_KEY = "id"19TEST_DDB_TABLE_NAME = "test-ddb-table-1"20TEST_DDB_TABLE_NAME_2 = "test-ddb-table-2"21TEST_DDB_TABLE_NAME_3 = "test-ddb-table-3"22TEST_DDB_TAGS = [23    {"Key": "Name", "Value": "test-table"},24    {"Key": "TestKey", "Value": "true"},25]26@pytest.fixture()27def dynamodb(dynamodb_resource):28    return dynamodb_resource29class TestDynamoDB:30    def test_non_ascii_chars(self, dynamodb):31        aws_stack.create_dynamodb_table(TEST_DDB_TABLE_NAME, partition_key=PARTITION_KEY)32        table = dynamodb.Table(TEST_DDB_TABLE_NAME)33        # write some items containing non-ASCII characters34        items = {35            "id1": {PARTITION_KEY: "id1", "data": "foobar123 â"},36            "id2": {PARTITION_KEY: "id2", "data": "foobar123 Â£"},37            "id3": {PARTITION_KEY: "id3", "data": "foobar123 Â¢"},38        }39        for k, item in items.items():40            table.put_item(Item=item)41        for item_id in items.keys():42            item = table.get_item(Key={PARTITION_KEY: item_id})["Item"]43            # need to fix up the JSON and convert str to unicode for Python 244            item1 = json_safe(item)45            item2 = json_safe(items[item_id])46            assert item1 == item247        # clean up48        delete_table(TEST_DDB_TABLE_NAME)49    def test_large_data_download(self, dynamodb):50        aws_stack.create_dynamodb_table(TEST_DDB_TABLE_NAME_2, partition_key=PARTITION_KEY)51        table = dynamodb.Table(TEST_DDB_TABLE_NAME_2)52        # Create a large amount of items53        num_items = 2054        for i in range(0, num_items):55            item = {PARTITION_KEY: "id%s" % i, "data1": "foobar123 " * 1000}56            table.put_item(Item=item)57        # Retrieve the items. The data will be transmitted to the client with chunked transfer encoding58        result = table.scan(TableName=TEST_DDB_TABLE_NAME_2)59        assert len(result["Items"]) == num_items60        # clean up61        delete_table(TEST_DDB_TABLE_NAME_2)62    def test_time_to_live(self, dynamodb):63        aws_stack.create_dynamodb_table(TEST_DDB_TABLE_NAME_3, partition_key=PARTITION_KEY)64        table = dynamodb.Table(TEST_DDB_TABLE_NAME_3)65        # Insert some items to the table66        items = {67            "id1": {PARTITION_KEY: "id1", "data": "IT IS"},68            "id2": {PARTITION_KEY: "id2", "data": "TIME"},69            "id3": {PARTITION_KEY: "id3", "data": "TO LIVE!"},70        }71        for k, item in items.items():72            table.put_item(Item=item)73        # Describe TTL when still unset74        response = testutil.send_describe_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3)75        assert response.status_code == 20076        assert (77            json.loads(response._content)["TimeToLiveDescription"]["TimeToLiveStatus"] == "DISABLED"78        )79        # Enable TTL for given table80        response = testutil.send_update_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3, True)81        assert response.status_code == 20082        assert json.loads(response._content)["TimeToLiveSpecification"]["Enabled"]83        # Describe TTL status after being enabled.84        response = testutil.send_describe_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3)85        assert response.status_code == 20086        assert (87            json.loads(response._content)["TimeToLiveDescription"]["TimeToLiveStatus"] == "ENABLED"88        )89        # Disable TTL for given table90        response = testutil.send_update_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3, False)91        assert response.status_code == 20092        assert not json.loads(response._content)["TimeToLiveSpecification"]["Enabled"]93        # Describe TTL status after being disabled.94        response = testutil.send_describe_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3)95        assert response.status_code == 20096        assert (97            json.loads(response._content)["TimeToLiveDescription"]["TimeToLiveStatus"] == "DISABLED"98        )99        # Enable TTL for given table again100        response = testutil.send_update_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3, True)101        assert response.status_code == 200102        assert json.loads(response._content)["TimeToLiveSpecification"]["Enabled"]103        # Describe TTL status after being enabled again.104        response = testutil.send_describe_dynamodb_ttl_request(TEST_DDB_TABLE_NAME_3)105        assert response.status_code == 200106        assert (107            json.loads(response._content)["TimeToLiveDescription"]["TimeToLiveStatus"] == "ENABLED"108        )109        # clean up110        delete_table(TEST_DDB_TABLE_NAME_3)111    def test_list_tags_of_resource(self, dynamodb):112        table_name = "ddb-table-%s" % short_uid()113        dynamodb = aws_stack.create_external_boto_client("dynamodb")114        rs = dynamodb.create_table(115            TableName=table_name,116            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],117            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],118            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},119            Tags=TEST_DDB_TAGS,120        )121        table_arn = rs["TableDescription"]["TableArn"]122        rs = dynamodb.list_tags_of_resource(ResourceArn=table_arn)123        assert rs["Tags"] == TEST_DDB_TAGS124        dynamodb.tag_resource(ResourceArn=table_arn, Tags=[{"Key": "NewKey", "Value": "TestValue"}])125        rs = dynamodb.list_tags_of_resource(ResourceArn=table_arn)126        assert len(rs["Tags"]) == len(TEST_DDB_TAGS) + 1127        tags = {tag["Key"]: tag["Value"] for tag in rs["Tags"]}128        assert "NewKey" in tags.keys()129        assert tags["NewKey"] == "TestValue"130        dynamodb.untag_resource(ResourceArn=table_arn, TagKeys=["Name", "NewKey"])131        rs = dynamodb.list_tags_of_resource(ResourceArn=table_arn)132        tags = {tag["Key"]: tag["Value"] for tag in rs["Tags"]}133        assert "Name" not in tags.keys()134        assert "NewKey" not in tags.keys()135        delete_table(table_name)136    def test_stream_spec_and_region_replacement(self, dynamodb):137        ddbstreams = aws_stack.create_external_boto_client("dynamodbstreams")138        kinesis = aws_stack.create_external_boto_client("kinesis")139        table_name = f"ddb-{short_uid()}"140        aws_stack.create_dynamodb_table(141            table_name,142            partition_key=PARTITION_KEY,143            stream_view_type="NEW_AND_OLD_IMAGES",144        )145        table = dynamodb.Table(table_name)146        # assert ARN formats147        expected_arn_prefix = "arn:aws:dynamodb:" + aws_stack.get_local_region()148        assert table.table_arn.startswith(expected_arn_prefix)149        assert table.latest_stream_arn.startswith(expected_arn_prefix)150        # assert stream has been created151        stream_tables = [s["TableName"] for s in ddbstreams.list_streams()["Streams"]]152        assert table_name in stream_tables153        stream_name = get_kinesis_stream_name(table_name)154        assert stream_name in kinesis.list_streams()["StreamNames"]155        # assert shard ID formats156        result = ddbstreams.describe_stream(StreamArn=table.latest_stream_arn)["StreamDescription"]157        assert "Shards" in result158        for shard in result["Shards"]:159            assert re.match(r"^shardId-[0-9]{20}-[a-zA-Z0-9]{1,36}$", shard["ShardId"])160        # clean up161        delete_table(table_name)162        def _assert_stream_deleted():163            stream_tables = [s["TableName"] for s in ddbstreams.list_streams()["Streams"]]164            assert table_name not in stream_tables165            assert stream_name not in kinesis.list_streams()["StreamNames"]166        # assert stream has been deleted167        retry(_assert_stream_deleted, sleep=0.4, retries=5)168    def test_multiple_update_expressions(self, dynamodb):169        dynamodb_client = aws_stack.create_external_boto_client("dynamodb")170        aws_stack.create_dynamodb_table(TEST_DDB_TABLE_NAME, partition_key=PARTITION_KEY)171        table = dynamodb.Table(TEST_DDB_TABLE_NAME)172        item_id = short_uid()173        table.put_item(Item={PARTITION_KEY: item_id, "data": "foobar123 â"})174        response = dynamodb_client.update_item(175            TableName=TEST_DDB_TABLE_NAME,176            Key={PARTITION_KEY: {"S": item_id}},177            UpdateExpression="SET attr1 = :v1, attr2 = :v2",178            ExpressionAttributeValues={":v1": {"S": "value1"}, ":v2": {"S": "value2"}},179        )180        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200181        item = table.get_item(Key={PARTITION_KEY: item_id})["Item"]182        assert item["attr1"] == "value1"183        assert item["attr2"] == "value2"184        attributes = [{"AttributeName": "id", "AttributeType": STRING}]185        user_id_idx = [186            {187                "Create": {188                    "IndexName": "id-index",189                    "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],190                    "Projection": {191                        "ProjectionType": "INCLUDE",192                        "NonKeyAttributes": ["data"],193                    },194                    "ProvisionedThroughput": {195                        "ReadCapacityUnits": 5,196                        "WriteCapacityUnits": 5,197                    },198                }199            },200        ]201        # for each index202        table.update(AttributeDefinitions=attributes, GlobalSecondaryIndexUpdates=user_id_idx)203        with pytest.raises(Exception) as ctx:204            table.query(205                TableName=TEST_DDB_TABLE_NAME,206                IndexName="id-index",207                KeyConditionExpression=Key(PARTITION_KEY).eq(item_id),208                Select="ALL_ATTRIBUTES",209            )210        assert ctx.match("ValidationException")211    def test_invalid_query_index(self, dynamodb):212        """Raises an exception when a query requests ALL_ATTRIBUTES,213        but the index does not have a ProjectionType of ALL"""214        table_name = f"test-table-{short_uid()}"215        table = dynamodb.create_table(216            TableName=table_name,217            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],218            AttributeDefinitions=[219                {"AttributeName": "id", "AttributeType": "S"},220                {"AttributeName": "field_a", "AttributeType": "S"},221            ],222            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},223            Tags=TEST_DDB_TAGS,224            GlobalSecondaryIndexes=[225                {226                    "IndexName": "field_a_index",227                    "KeySchema": [{"AttributeName": "field_a", "KeyType": "HASH"}],228                    "Projection": {"ProjectionType": "KEYS_ONLY"},229                    "ProvisionedThroughput": {230                        "ReadCapacityUnits": 1,231                        "WriteCapacityUnits": 1,232                    },233                },234            ],235        )236        with pytest.raises(Exception) as ctx:237            table.query(238                TableName=table_name,239                IndexName="field_a_index",240                KeyConditionExpression=Key("field_a").eq("xyz"),241                Select="ALL_ATTRIBUTES",242            )243        assert ctx.match("ValidationException")244        # clean up245        delete_table(table_name)246    def test_valid_query_index(self, dynamodb):247        """Query requests ALL_ATTRIBUTES and the named index has a ProjectionType of ALL,248        no exception should be raised."""249        table_name = f"test-table-{short_uid()}"250        table = dynamodb.create_table(251            TableName=table_name,252            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],253            AttributeDefinitions=[254                {"AttributeName": "id", "AttributeType": "S"},255                {"AttributeName": "field_a", "AttributeType": "S"},256                {"AttributeName": "field_b", "AttributeType": "S"},257            ],258            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},259            Tags=TEST_DDB_TAGS,260            GlobalSecondaryIndexes=[261                {262                    "IndexName": "field_a_index",263                    "KeySchema": [{"AttributeName": "field_a", "KeyType": "HASH"}],264                    "Projection": {"ProjectionType": "KEYS_ONLY"},265                    "ProvisionedThroughput": {266                        "ReadCapacityUnits": 1,267                        "WriteCapacityUnits": 1,268                    },269                },270                {271                    "IndexName": "field_b_index",272                    "KeySchema": [{"AttributeName": "field_b", "KeyType": "HASH"}],273                    "Projection": {"ProjectionType": "ALL"},274                    "ProvisionedThroughput": {275                        "ReadCapacityUnits": 1,276                        "WriteCapacityUnits": 1,277                    },278                },279            ],280        )281        table.query(282            TableName=table_name,283            IndexName="field_b_index",284            KeyConditionExpression=Key("field_b").eq("xyz"),285            Select="ALL_ATTRIBUTES",286        )287        # clean up288        delete_table(table_name)289    def test_valid_local_secondary_index(290        self, dynamodb_client, dynamodb_create_table_with_parameters, dynamodb_wait_for_table_active291    ):292        try:293            table_name = f"test-table-{short_uid()}"294            dynamodb_create_table_with_parameters(295                TableName=table_name,296                KeySchema=[297                    {"AttributeName": "PK", "KeyType": "HASH"},298                    {"AttributeName": "SK", "KeyType": "RANGE"},299                ],300                AttributeDefinitions=[301                    {"AttributeName": "PK", "AttributeType": "S"},302                    {"AttributeName": "SK", "AttributeType": "S"},303                    {"AttributeName": "LSI1SK", "AttributeType": "N"},304                ],305                LocalSecondaryIndexes=[306                    {307                        "IndexName": "LSI1",308                        "KeySchema": [309                            {"AttributeName": "PK", "KeyType": "HASH"},310                            {"AttributeName": "LSI1SK", "KeyType": "RANGE"},311                        ],312                        "Projection": {"ProjectionType": "ALL"},313                    }314                ],315                ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},316                Tags=TEST_DDB_TAGS,317            )318            dynamodb_wait_for_table_active(table_name)319            item = {"SK": {"S": "hello"}, "LSI1SK": {"N": "123"}, "PK": {"S": "test one"}}320            dynamodb_client.put_item(TableName=table_name, Item=item)321            result = dynamodb_client.query(322                TableName=table_name,323                IndexName="LSI1",324                KeyConditionExpression="PK = :v1",325                ExpressionAttributeValues={":v1": {"S": "test one"}},326                Select="ALL_ATTRIBUTES",327            )328            assert result["Items"] == [item]329        finally:330            dynamodb_client.delete_table(TableName=table_name)331    def test_more_than_20_global_secondary_indexes(self, dynamodb, dynamodb_client):332        table_name = f"test-table-{short_uid()}"333        num_gsis = 25334        attrs = [{"AttributeName": f"a{i}", "AttributeType": "S"} for i in range(num_gsis)]335        gsis = [336            {337                "IndexName": f"gsi_{i}",338                "KeySchema": [{"AttributeName": f"a{i}", "KeyType": "HASH"}],339                "Projection": {"ProjectionType": "ALL"},340            }341            for i in range(num_gsis)342        ]343        dynamodb.create_table(344            TableName=table_name,345            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],346            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}, *attrs],347            GlobalSecondaryIndexes=gsis,348            BillingMode="PAY_PER_REQUEST",349        )350        table = dynamodb_client.describe_table(TableName=table_name)351        assert len(table["Table"]["GlobalSecondaryIndexes"]) == num_gsis352        # clean up353        delete_table(table_name)354    @pytest.mark.aws_validated355    def test_return_values_in_put_item(self, dynamodb, dynamodb_client):356        aws_stack.create_dynamodb_table(357            TEST_DDB_TABLE_NAME, partition_key=PARTITION_KEY, client=dynamodb_client358        )359        table = dynamodb.Table(TEST_DDB_TABLE_NAME)360        def _validate_response(response, expected: dict = {}):361            """362            Validates the response against the optionally expected one.363            It checks that the response doesn't contain `Attributes`,364            `ConsumedCapacity` and `ItemCollectionMetrics` unless they are expected.365            """366            should_not_contain = {367                "Attributes",368                "ConsumedCapacity",369                "ItemCollectionMetrics",370            } - expected.keys()371            assert response["ResponseMetadata"]["HTTPStatusCode"] == 200372            assert expected.items() <= response.items()373            assert response.keys().isdisjoint(should_not_contain)374        # items which are being used to put in the table375        item1 = {PARTITION_KEY: "id1", "data": "foobar"}376        item1b = {PARTITION_KEY: "id1", "data": "barfoo"}377        item2 = {PARTITION_KEY: "id2", "data": "foobar"}378        response = table.put_item(Item=item1, ReturnValues="ALL_OLD")379        # there is no data present in the table already so even if return values380        # is set to 'ALL_OLD' as there is no data it will not return any data.381        _validate_response(response)382        # now the same data is present so when we pass return values as 'ALL_OLD'383        # it should give us attributes384        response = table.put_item(Item=item1, ReturnValues="ALL_OLD")385        _validate_response(response, expected={"Attributes": item1})386        # now a previous version of data is present, so when we pass return387        # values as 'ALL_OLD' it should give us the old attributes388        response = table.put_item(Item=item1b, ReturnValues="ALL_OLD")389        _validate_response(response, expected={"Attributes": item1})390        response = table.put_item(Item=item2)391        # we do not have any same item as item2 already so when we add this by default392        # return values is set to None so no Attribute values should be returned393        _validate_response(response)394        response = table.put_item(Item=item2)395        # in this case we already have item2 in the table so on this request396        # it should not return any data as return values is set to None so no397        # Attribute values should be returned398        _validate_response(response)399        # cleanup400        table.delete()401    @pytest.mark.aws_validated402    def test_empty_and_binary_values(self, dynamodb, dynamodb_client):403        aws_stack.create_dynamodb_table(404            TEST_DDB_TABLE_NAME, partition_key=PARTITION_KEY, client=dynamodb_client405        )406        table = dynamodb.Table(TEST_DDB_TABLE_NAME)407        # items which are being used to put in the table408        item1 = {PARTITION_KEY: "id1", "data": ""}409        item2 = {PARTITION_KEY: "id2", "data": b"\x90"}410        response = table.put_item(Item=item1)411        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200412        response = table.put_item(Item=item2)413        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200414        # clean up415        table.delete()416    def test_batch_write_binary(self, dynamodb_client):417        table_name = "table_batch_binary_%s" % short_uid()418        dynamodb_client.create_table(419            TableName=table_name,420            AttributeDefinitions=[421                {"AttributeName": "PK", "AttributeType": "S"},422                {"AttributeName": "SK", "AttributeType": "S"},423            ],424            KeySchema=[425                {"AttributeName": "PK", "KeyType": "HASH"},426                {"AttributeName": "SK", "KeyType": "RANGE"},427            ],428            BillingMode="PAY_PER_REQUEST",429        )430        dynamodb_client.put_item(431            TableName=table_name,432            Item={"PK": {"S": "hello"}, "SK": {"S": "user"}, "data": {"B": b"test"}},433        )434        item = {435            "Item": {436                "PK": {"S": "hello-1"},437                "SK": {"S": "user-1"},438                "data": {"B": b"test-1"},439            }440        }441        item_non_decodable = {442            "Item": {443                "PK": {"S": "hello-2"},444                "SK": {"S": "user-2"},445                "data": {"B": b"test \xc0 \xed"},446            }447        }448        response = dynamodb_client.batch_write_item(449            RequestItems={table_name: [{"PutRequest": item}, {"PutRequest": item_non_decodable}]}450        )451        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200452        dynamodb_client.delete_table(TableName=table_name)453    def test_binary_data_with_stream(454        self,455        wait_for_stream_ready,456        dynamodb_create_table_with_parameters,457        dynamodb_client,458        kinesis_client,459    ):460        table_name = f"table-{short_uid()}"461        dynamodb_create_table_with_parameters(462            TableName=table_name,463            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],464            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],465            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},466            StreamSpecification={467                "StreamEnabled": True,468                "StreamViewType": "NEW_AND_OLD_IMAGES",469            },470        )471        stream_name = get_kinesis_stream_name(table_name)472        wait_for_stream_ready(stream_name)473        response = dynamodb_client.put_item(474            TableName=table_name, Item={"id": {"S": "id1"}, "data": {"B": b"\x90"}}475        )476        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200477        iterator = get_shard_iterator(stream_name, kinesis_client)478        response = kinesis_client.get_records(ShardIterator=iterator)479        json_records = response.get("Records")480        assert 1 == len(json_records)481        assert "Data" in json_records[0]482    def test_dynamodb_stream_shard_iterator(self, wait_for_stream_ready):483        dynamodb = aws_stack.create_external_boto_client("dynamodb")484        ddbstreams = aws_stack.create_external_boto_client("dynamodbstreams")485        table_name = "table_with_stream-%s" % short_uid()486        table = dynamodb.create_table(487            TableName=table_name,488            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],489            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],490            StreamSpecification={491                "StreamEnabled": True,492                "StreamViewType": "NEW_IMAGE",493            },494            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},495        )496        stream_name = get_kinesis_stream_name(table_name)497        wait_for_stream_ready(stream_name)498        stream_arn = table["TableDescription"]["LatestStreamArn"]499        result = ddbstreams.describe_stream(StreamArn=stream_arn)500        response = ddbstreams.get_shard_iterator(501            StreamArn=stream_arn,502            ShardId=result["StreamDescription"]["Shards"][0]["ShardId"],503            ShardIteratorType="LATEST",504        )505        assert "ShardIterator" in response506        response = ddbstreams.get_shard_iterator(507            StreamArn=stream_arn,508            ShardId=result["StreamDescription"]["Shards"][0]["ShardId"],509            ShardIteratorType="AT_SEQUENCE_NUMBER",510            SequenceNumber=result["StreamDescription"]["Shards"][0]511            .get("SequenceNumberRange")512            .get("StartingSequenceNumber"),513        )514        assert "ShardIterator" in response515    def test_dynamodb_create_table_with_class(self, dynamodb_client):516        table_name = "table_with_class_%s" % short_uid()517        # create table518        result = dynamodb_client.create_table(519            TableName=table_name,520            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],521            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],522            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},523            TableClass="STANDARD",524        )525        assert result["TableDescription"]["TableClassSummary"]["TableClass"] == "STANDARD"526        result = dynamodb_client.describe_table(TableName=table_name)527        assert result["Table"]["TableClassSummary"]["TableClass"] == "STANDARD"528        result = dynamodb_client.update_table(529            TableName=table_name, TableClass="STANDARD_INFREQUENT_ACCESS"530        )531        assert (532            result["TableDescription"]["TableClassSummary"]["TableClass"]533            == "STANDARD_INFREQUENT_ACCESS"534        )535        result = dynamodb_client.describe_table(TableName=table_name)536        assert result["Table"]["TableClassSummary"]["TableClass"] == "STANDARD_INFREQUENT_ACCESS"537        # clean resources538        dynamodb_client.delete_table(TableName=table_name)539    def test_dynamodb_execute_transaction(self, dynamodb_client):540        table_name = "table_%s" % short_uid()541        dynamodb_client.create_table(542            TableName=table_name,543            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],544            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],545            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},546        )547        statements = [548            {"Statement": f"INSERT INTO {table_name} VALUE {{'Username': 'user01'}}"},549            {"Statement": f"INSERT INTO {table_name} VALUE {{'Username': 'user02'}}"},550        ]551        result = dynamodb_client.execute_transaction(TransactStatements=statements)552        assert result["ResponseMetadata"]["HTTPStatusCode"] == 200553        result = dynamodb_client.scan(TableName=table_name)554        assert result["ScannedCount"] == 2555        dynamodb_client.delete_table(TableName=table_name)556    def test_dynamodb_batch_execute_statement(self, dynamodb_client):557        table_name = "table_%s" % short_uid()558        dynamodb_client.create_table(559            TableName=table_name,560            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],561            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],562            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},563        )564        dynamodb_client.put_item(TableName=table_name, Item={"Username": {"S": "user02"}})565        statements = [566            {"Statement": f"INSERT INTO {table_name} VALUE {{'Username': 'user01'}}"},567            {"Statement": f"UPDATE {table_name} SET Age=20 WHERE Username='user02'"},568        ]569        result = dynamodb_client.batch_execute_statement(Statements=statements)570        # actions always succeeds571        assert not any("Error" in r for r in result["Responses"])572        item = dynamodb_client.get_item(TableName=table_name, Key={"Username": {"S": "user02"}})[573            "Item"574        ]575        assert item["Age"]["N"] == "20"576        item = dynamodb_client.get_item(TableName=table_name, Key={"Username": {"S": "user01"}})[577            "Item"578        ]579        assert item580        dynamodb_client.delete_table(TableName=table_name)581    def test_dynamodb_partiql_missing(self, dynamodb_client):582        table_name = "table_with_stream_%s" % short_uid()583        # create table584        dynamodb_client.create_table(585            TableName=table_name,586            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],587            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],588            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},589        )590        # create items with FirstName attribute591        dynamodb_client.execute_statement(592            Statement=f"INSERT INTO {table_name} VALUE {{'Username': 'Alice123', 'FirstName':'Alice'}}"593        )594        items = dynamodb_client.execute_statement(595            Statement=f"SELECT * FROM {table_name} WHERE FirstName IS NOT MISSING"596        )["Items"]597        assert len(items) == 1598        items = dynamodb_client.execute_statement(599            Statement=f"SELECT * FROM {table_name} WHERE FirstName IS MISSING"600        )["Items"]601        assert len(items) == 0602        dynamodb_client.delete_table(TableName=table_name)603    def test_dynamodb_stream_stream_view_type(self):604        dynamodb = aws_stack.create_external_boto_client("dynamodb")605        ddbstreams = aws_stack.create_external_boto_client("dynamodbstreams")606        table_name = "table_with_stream_%s" % short_uid()607        # create table608        table = dynamodb.create_table(609            TableName=table_name,610            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],611            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],612            StreamSpecification={613                "StreamEnabled": True,614                "StreamViewType": "KEYS_ONLY",615            },616            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},617        )618        stream_arn = table["TableDescription"]["LatestStreamArn"]619        # wait for stream to be created620        sleep(1)621        # put item in table - INSERT event622        dynamodb.put_item(TableName=table_name, Item={"Username": {"S": "Fred"}})623        # update item in table - MODIFY event624        dynamodb.update_item(625            TableName=table_name,626            Key={"Username": {"S": "Fred"}},627            UpdateExpression="set S=:r",628            ExpressionAttributeValues={":r": {"S": "Fred_Modified"}},629            ReturnValues="UPDATED_NEW",630        )631        # delete item in table - REMOVE event632        dynamodb.delete_item(TableName=table_name, Key={"Username": {"S": "Fred"}})633        result = ddbstreams.describe_stream(StreamArn=stream_arn)634        # assert stream_view_type of the table635        assert result["StreamDescription"]["StreamViewType"] == "KEYS_ONLY"636        # add item via PartiQL query - INSERT event637        dynamodb.execute_statement(638            Statement=f"INSERT INTO {table_name} VALUE {{'Username': 'Alice'}}"639        )640        # run update via PartiQL query - MODIFY event641        dynamodb.execute_statement(642            Statement=f"UPDATE {table_name} SET partiql=1 WHERE Username='Alice'"643        )644        # run update via PartiQL query - REMOVE event645        dynamodb.execute_statement(Statement=f"DELETE FROM {table_name} WHERE Username='Alice'")646        # get shard iterator647        response = ddbstreams.get_shard_iterator(648            StreamArn=stream_arn,649            ShardId=result["StreamDescription"]["Shards"][0]["ShardId"],650            ShardIteratorType="AT_SEQUENCE_NUMBER",651            SequenceNumber=result["StreamDescription"]["Shards"][0]652            .get("SequenceNumberRange")653            .get("StartingSequenceNumber"),654        )655        # get stream records656        records = ddbstreams.get_records(ShardIterator=response["ShardIterator"])["Records"]657        assert len(records) == 6658        events = [rec["eventName"] for rec in records]659        assert events == ["INSERT", "MODIFY", "REMOVE"] * 2660        # assert that all records contain proper event IDs661        event_ids = [rec.get("eventID") for rec in records]662        assert all(event_ids)663        # assert that updates have been received from regular table operations and PartiQL query operations664        for idx, record in enumerate(records):665            assert "SequenceNumber" in record["dynamodb"]666            assert record["dynamodb"]["StreamViewType"] == "KEYS_ONLY"667            assert record["dynamodb"]["Keys"] == {"Username": {"S": "Fred" if idx < 3 else "Alice"}}668            assert "OldImage" not in record["dynamodb"]669            assert "NewImage" not in record["dynamodb"]670        # clean up671        delete_table(table_name)672    def test_dynamodb_with_kinesis_stream(self):673        dynamodb = aws_stack.create_external_boto_client("dynamodb")674        kinesis = aws_stack.create_external_boto_client("kinesis")675        # create kinesis datastream676        stream_name = "kinesis_dest_stream"677        kinesis.create_stream(StreamName=stream_name, ShardCount=1)678        # wait for the stream to be created679        sleep(1)680        # Get stream description681        stream_description = kinesis.describe_stream(StreamName=stream_name)["StreamDescription"]682        table_name = "table_with_kinesis_stream-%s" % short_uid()683        # create table684        dynamodb.create_table(685            TableName=table_name,686            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],687            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],688            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},689        )690        # Enable kinesis destination for the table691        dynamodb.enable_kinesis_streaming_destination(692            TableName=table_name, StreamArn=stream_description["StreamARN"]693        )694        # put item into table695        dynamodb.put_item(TableName=table_name, Item={"Username": {"S": "Fred"}})696        # update item in table697        dynamodb.update_item(698            TableName=table_name,699            Key={"Username": {"S": "Fred"}},700            UpdateExpression="set S=:r",701            ExpressionAttributeValues={":r": {"S": "Fred_Modified"}},702            ReturnValues="UPDATED_NEW",703        )704        # delete item in table705        dynamodb.delete_item(TableName=table_name, Key={"Username": {"S": "Fred"}})706        def _fetch_records():707            records = aws_stack.kinesis_get_latest_records(708                stream_name, shard_id=stream_description["Shards"][0]["ShardId"]709            )710            assert len(records) == 3711            return records712        # get records from the stream713        records = retry(_fetch_records)714        for record in records:715            record = json.loads(record["Data"])716            assert record["tableName"] == table_name717            # check eventSourceARN not exists in the stream record718            assert "eventSourceARN" not in record719            if record["eventName"] == "INSERT":720                assert "OldImage" not in record["dynamodb"]721                assert "NewImage" in record["dynamodb"]722            elif record["eventName"] == "MODIFY":723                assert "NewImage" in record["dynamodb"]724                assert "OldImage" in record["dynamodb"]725            elif record["eventName"] == "REMOVE":726                assert "NewImage" not in record["dynamodb"]727                assert "OldImage" in record["dynamodb"]728        # describe kinesis streaming destination of the table729        destinations = dynamodb.describe_kinesis_streaming_destination(TableName=table_name)730        destination = destinations["KinesisDataStreamDestinations"][0]731        # assert kinesis streaming destination status732        assert stream_description["StreamARN"] == destination["StreamArn"]733        assert destination["DestinationStatus"] == "ACTIVE"734        # Disable kinesis destination735        dynamodb.disable_kinesis_streaming_destination(736            TableName=table_name, StreamArn=stream_description["StreamARN"]737        )738        # describe kinesis streaming destination of the table739        result = dynamodb.describe_kinesis_streaming_destination(TableName=table_name)740        destination = result["KinesisDataStreamDestinations"][0]741        # assert kinesis streaming destination status742        assert stream_description["StreamARN"] == destination["StreamArn"]743        assert destination["DestinationStatus"] == "DISABLED"744        # clean up745        delete_table(table_name)746        kinesis.delete_stream(StreamName="kinesis_dest_stream")747    def test_global_tables(self):748        aws_stack.create_dynamodb_table(TEST_DDB_TABLE_NAME, partition_key=PARTITION_KEY)749        dynamodb = aws_stack.create_external_boto_client("dynamodb")750        # create global table751        regions = [752            {"RegionName": "us-east-1"},753            {"RegionName": "us-west-1"},754            {"RegionName": "eu-central-1"},755        ]756        response = dynamodb.create_global_table(757            GlobalTableName=TEST_DDB_TABLE_NAME, ReplicationGroup=regions758        )["GlobalTableDescription"]759        assert "ReplicationGroup" in response760        assert len(response["ReplicationGroup"]) == len(regions)761        # describe global table762        response = dynamodb.describe_global_table(GlobalTableName=TEST_DDB_TABLE_NAME)[763            "GlobalTableDescription"764        ]765        assert "ReplicationGroup" in response766        assert len(regions) == len(response["ReplicationGroup"])767        # update global table768        updates = [769            {"Create": {"RegionName": "us-east-2"}},770            {"Create": {"RegionName": "us-west-2"}},771            {"Delete": {"RegionName": "us-west-1"}},772        ]773        response = dynamodb.update_global_table(774            GlobalTableName=TEST_DDB_TABLE_NAME, ReplicaUpdates=updates775        )["GlobalTableDescription"]776        assert "ReplicationGroup" in response777        assert len(response["ReplicationGroup"]) == len(regions) + 1778        # assert exceptions for invalid requests779        with pytest.raises(Exception) as ctx:780            dynamodb.create_global_table(781                GlobalTableName=TEST_DDB_TABLE_NAME, ReplicationGroup=regions782            )783        assert ctx.match("GlobalTableAlreadyExistsException")784        with pytest.raises(Exception) as ctx:785            dynamodb.describe_global_table(GlobalTableName="invalid-table-name")786        assert ctx.match("GlobalTableNotFoundException")787    def test_create_duplicate_table(self, dynamodb_create_table_with_parameters):788        table_name = "duplicateTable"789        dynamodb_create_table_with_parameters(790            TableName=table_name,791            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],792            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],793            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},794            Tags=TEST_DDB_TAGS,795        )796        with pytest.raises(Exception) as ctx:797            dynamodb_create_table_with_parameters(798                TableName=table_name,799                KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],800                AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],801                ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},802                Tags=TEST_DDB_TAGS,803            )804        ctx.match("ResourceInUseException")805    def test_delete_table(self, dynamodb_client, dynamodb_create_table):806        table_name = "test-ddb-table-%s" % short_uid()807        tables_before = len(dynamodb_client.list_tables()["TableNames"])808        dynamodb_create_table(809            table_name=table_name,810            partition_key=PARTITION_KEY,811        )812        table_list = dynamodb_client.list_tables()813        # TODO: fix assertion, to enable parallel test execution!814        assert tables_before + 1 == len(table_list["TableNames"])815        assert table_name in table_list["TableNames"]816        dynamodb_client.delete_table(TableName=table_name)817        table_list = dynamodb_client.list_tables()818        assert tables_before == len(table_list["TableNames"])819        with pytest.raises(Exception) as ctx:820            dynamodb_client.delete_table(TableName=table_name)821        assert ctx.match("ResourceNotFoundException")822    def test_transaction_write_items(self, dynamodb_client, dynamodb_create_table_with_parameters):823        table_name = "test-ddb-table-%s" % short_uid()824        dynamodb_create_table_with_parameters(825            TableName=table_name,826            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],827            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],828            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},829            Tags=TEST_DDB_TAGS,830        )831        response = dynamodb_client.transact_write_items(832            TransactItems=[833                {834                    "ConditionCheck": {835                        "TableName": table_name,836                        "ConditionExpression": "attribute_not_exists(id)",837                        "Key": {"id": {"S": "test1"}},838                    }839                },840                {"Put": {"TableName": table_name, "Item": {"id": {"S": "test2"}}}},841                {842                    "Update": {843                        "TableName": table_name,844                        "Key": {"id": {"S": "test3"}},845                        "UpdateExpression": "SET attr1 = :v1, attr2 = :v2",846                        "ExpressionAttributeValues": {847                            ":v1": {"S": "value1"},848                            ":v2": {"S": "value2"},849                        },850                    }851                },852                {"Delete": {"TableName": table_name, "Key": {"id": {"S": "test4"}}}},853            ]854        )855        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200856    @pytest.mark.aws_validated857    def test_transaction_write_canceled(858        self, dynamodb_create_table_with_parameters, dynamodb_wait_for_table_active, dynamodb_client859    ):860        table_name = "table_%s" % short_uid()861        # create table862        dynamodb_create_table_with_parameters(863            TableName=table_name,864            KeySchema=[{"AttributeName": "Username", "KeyType": "HASH"}],865            AttributeDefinitions=[{"AttributeName": "Username", "AttributeType": "S"}],866            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},867        )868        dynamodb_wait_for_table_active(table_name)869        # put item in table - INSERT event870        dynamodb_client.put_item(TableName=table_name, Item={"Username": {"S": "Fred"}})871        # provoke a TransactionCanceledException by adding a condition which is not met872        with pytest.raises(Exception) as ctx:873            dynamodb_client.transact_write_items(874                TransactItems=[875                    {876                        "ConditionCheck": {877                            "TableName": table_name,878                            "ConditionExpression": "attribute_not_exists(Username)",879                            "Key": {"Username": {"S": "Fred"}},880                        }881                    },882                    {"Delete": {"TableName": table_name, "Key": {"Username": {"S": "Bert"}}}},883                ]884            )885        # Make sure the exception contains the cancellation reasons886        assert ctx.match("TransactionCanceledException")887        assert (888            str(ctx.value)889            == "An error occurred (TransactionCanceledException) when calling the TransactWriteItems operation: "890            "Transaction cancelled, please refer cancellation reasons for specific reasons "891            "[ConditionalCheckFailed, None]"892        )893        assert hasattr(ctx.value, "response")894        assert "CancellationReasons" in ctx.value.response895        conditional_check_failed = [896            reason897            for reason in ctx.value.response["CancellationReasons"]898            if reason.get("Code") == "ConditionalCheckFailed"899        ]900        assert len(conditional_check_failed) == 1901        assert "Message" in conditional_check_failed[0]902        # dynamodb-local adds a trailing "." to the message, AWS does not903        assert re.match(904            r"^The conditional request failed\.?$", conditional_check_failed[0]["Message"]905        )906    def test_transaction_write_binary_data(907        self, dynamodb_client, dynamodb_create_table_with_parameters908    ):909        table_name = "test-ddb-table-%s" % short_uid()910        dynamodb_create_table_with_parameters(911            TableName=table_name,912            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],913            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],914            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},915            Tags=TEST_DDB_TAGS,916        )917        binary_item = {"B": b"foobar"}918        response = dynamodb_client.transact_write_items(919            TransactItems=[920                {921                    "Put": {922                        "TableName": table_name,923                        "Item": {924                            "id": {"S": "someUser"},925                            "binaryData": binary_item,926                        },927                    }928                }929            ]930        )931        item = dynamodb_client.get_item(TableName=table_name, Key={"id": {"S": "someUser"}})["Item"]932        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200933        assert item["binaryData"]934        assert item["binaryData"] == binary_item935    def test_transact_get_items(self, dynamodb_client, dynamodb_create_table):936        table_name = "test-ddb-table-%s" % short_uid()937        dynamodb_create_table(938            table_name=table_name,939            partition_key=PARTITION_KEY,940        )941        dynamodb_client.put_item(TableName=table_name, Item={"id": {"S": "John"}})942        result = dynamodb_client.transact_get_items(943            TransactItems=[{"Get": {"Key": {"id": {"S": "John"}}, "TableName": table_name}}]944        )945        assert result["ResponseMetadata"]["HTTPStatusCode"] == 200946    def test_batch_write_items(self, dynamodb_client, dynamodb_create_table_with_parameters):947        table_name = "test-ddb-table-%s" % short_uid()948        dynamodb_create_table_with_parameters(949            TableName=table_name,950            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],951            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],952            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},953            Tags=TEST_DDB_TAGS,954        )955        dynamodb_client.put_item(TableName=table_name, Item={"id": {"S": "Fred"}})956        response = dynamodb_client.batch_write_item(957            RequestItems={958                table_name: [959                    {"DeleteRequest": {"Key": {"id": {"S": "Fred"}}}},960                    {"PutRequest": {"Item": {"id": {"S": "Bob"}}}},961                ]962            }963        )964        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200965    @pytest.mark.xfail(reason="this test flakes regularly in CI")966    def test_dynamodb_stream_records_with_update_item(967        self,968        dynamodb_client,969        dynamodbstreams_client,970        dynamodb_resource,971        dynamodb_create_table,972        wait_for_stream_ready,973    ):974        table_name = f"test-ddb-table-{short_uid()}"975        dynamodb_create_table(976            table_name=table_name,977            partition_key=PARTITION_KEY,978            stream_view_type="NEW_AND_OLD_IMAGES",979        )980        table = dynamodb_resource.Table(table_name)981        stream_name = get_kinesis_stream_name(table_name)982        wait_for_stream_ready(stream_name)983        response = dynamodbstreams_client.describe_stream(StreamArn=table.latest_stream_arn)984        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200985        assert len(response["StreamDescription"]["Shards"]) == 1986        shard_id = response["StreamDescription"]["Shards"][0]["ShardId"]987        starting_sequence_number = int(988            response["StreamDescription"]["Shards"][0]989            .get("SequenceNumberRange")990            .get("StartingSequenceNumber")991        )992        response = dynamodbstreams_client.get_shard_iterator(993            StreamArn=table.latest_stream_arn,994            ShardId=shard_id,995            ShardIteratorType="LATEST",996        )997        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200998        assert "ShardIterator" in response999        iterator_id = response["ShardIterator"]1000        item_id = short_uid()1001        for _ in range(2):1002            dynamodb_client.update_item(1003                TableName=table_name,1004                Key={PARTITION_KEY: {"S": item_id}},1005                UpdateExpression="SET attr1 = :v1, attr2 = :v2",1006                ExpressionAttributeValues={1007                    ":v1": {"S": "value1"},1008                    ":v2": {"S": "value2"},1009                },1010                ReturnValues="ALL_NEW",1011                ReturnConsumedCapacity="INDEXES",1012            )1013        def check_expected_records():1014            records = dynamodbstreams_client.get_records(ShardIterator=iterator_id)1015            assert records["ResponseMetadata"]["HTTPStatusCode"] == 2001016            assert len(records["Records"]) == 21017            assert isinstance(1018                records["Records"][0]["dynamodb"]["ApproximateCreationDateTime"],1019                datetime,1020            )1021            assert records["Records"][0]["dynamodb"]["ApproximateCreationDateTime"].microsecond == 01022            assert records["Records"][0]["eventVersion"] == "1.1"1023            assert records["Records"][0]["eventName"] == "INSERT"1024            assert "OldImage" not in records["Records"][0]["dynamodb"]1025            assert (1026                int(records["Records"][0]["dynamodb"]["SequenceNumber"]) > starting_sequence_number1027            )1028            assert isinstance(1029                records["Records"][1]["dynamodb"]["ApproximateCreationDateTime"],1030                datetime,1031            )1032            assert records["Records"][1]["dynamodb"]["ApproximateCreationDateTime"].microsecond == 01033            assert records["Records"][1]["eventVersion"] == "1.1"1034            assert records["Records"][1]["eventName"] == "MODIFY"1035            assert "OldImage" in records["Records"][1]["dynamodb"]1036            assert (1037                int(records["Records"][1]["dynamodb"]["SequenceNumber"]) > starting_sequence_number1038            )1039        retry(check_expected_records, retries=5, sleep=1, sleep_before=2)1040    def test_query_on_deleted_resource(self, dynamodb_client, dynamodb_create_table):1041        table_name = "ddb-table-%s" % short_uid()1042        partition_key = "username"1043        dynamodb_create_table(table_name=table_name, partition_key=partition_key)1044        rs = dynamodb_client.query(1045            TableName=table_name,1046            KeyConditionExpression="{} = :username".format(partition_key),1047            ExpressionAttributeValues={":username": {"S": "test"}},1048        )1049        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 2001050        dynamodb_client.delete_table(TableName=table_name)1051        with pytest.raises(Exception) as ctx:1052            dynamodb_client.query(1053                TableName=table_name,1054                KeyConditionExpression="{} = :username".format(partition_key),1055                ExpressionAttributeValues={":username": {"S": "test"}},1056            )1057        assert ctx.match("ResourceNotFoundException")1058    def test_dynamodb_stream_to_lambda(1059        self, lambda_client, dynamodb_resource, dynamodb_create_table, wait_for_stream_ready1060    ):1061        table_name = "ddb-table-%s" % short_uid()1062        function_name = "func-%s" % short_uid()1063        partition_key = "SK"1064        dynamodb_create_table(1065            table_name=table_name,1066            partition_key=partition_key,1067            stream_view_type="NEW_AND_OLD_IMAGES",1068        )1069        table = dynamodb_resource.Table(table_name)1070        latest_stream_arn = table.latest_stream_arn1071        stream_name = get_kinesis_stream_name(table_name)1072        wait_for_stream_ready(stream_name)1073        testutil.create_lambda_function(1074            handler_file=TEST_LAMBDA_PYTHON_ECHO,1075            func_name=function_name,1076            runtime=LAMBDA_RUNTIME_PYTHON36,1077        )1078        mapping_uuid = lambda_client.create_event_source_mapping(1079            EventSourceArn=latest_stream_arn,1080            FunctionName=function_name,1081            StartingPosition="TRIM_HORIZON",1082        )["UUID"]1083        item = {"SK": short_uid(), "Name": "name-{}".format(short_uid())}1084        table.put_item(Item=item)1085        events = retry(1086            check_expected_lambda_log_events_length,1087            retries=10,1088            sleep=1,1089            function_name=function_name,1090            expected_length=1,1091            regex_filter=r"Records",1092        )1093        assert len(events) == 11094        assert len(events[0]["Records"]) == 11095        dynamodb_event = events[0]["Records"][0]["dynamodb"]1096        assert dynamodb_event["StreamViewType"] == "NEW_AND_OLD_IMAGES"1097        assert dynamodb_event["Keys"] == {"SK": {"S": item["SK"]}}1098        assert dynamodb_event["NewImage"]["Name"] == {"S": item["Name"]}1099        assert "SequenceNumber" in dynamodb_event1100        lambda_client.delete_event_source_mapping(UUID=mapping_uuid)1101    def test_dynamodb_batch_write_item(1102        self, dynamodb_client, dynamodb_create_table_with_parameters1103    ):1104        table_name = "ddb-table-%s" % short_uid()1105        dynamodb_create_table_with_parameters(1106            TableName=table_name,1107            KeySchema=[{"AttributeName": PARTITION_KEY, "KeyType": "HASH"}],1108            AttributeDefinitions=[{"AttributeName": PARTITION_KEY, "AttributeType": "S"}],1109            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},1110            Tags=TEST_DDB_TAGS,1111        )1112        result = dynamodb_client.batch_write_item(1113            RequestItems={1114                table_name: [1115                    {"PutRequest": {"Item": {PARTITION_KEY: {"S": "Test1"}}}},1116                    {"PutRequest": {"Item": {PARTITION_KEY: {"S": "Test2"}}}},1117                    {"PutRequest": {"Item": {PARTITION_KEY: {"S": "Test3"}}}},1118                ]1119            }1120        )1121        assert result.get("UnprocessedItems") == {}1122    def test_dynamodb_pay_per_request(self, dynamodb_create_table_with_parameters):1123        table_name = "ddb-table-%s" % short_uid()1124        with pytest.raises(Exception) as e:1125            dynamodb_create_table_with_parameters(1126                TableName=table_name,1127                KeySchema=[{"AttributeName": PARTITION_KEY, "KeyType": "HASH"}],1128                AttributeDefinitions=[{"AttributeName": PARTITION_KEY, "AttributeType": "S"}],1129                ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},1130                BillingMode="PAY_PER_REQUEST",1131            )1132        assert e.match("ValidationException")1133    def test_dynamodb_create_table_with_sse_specification(1134        self, dynamodb_create_table_with_parameters1135    ):1136        table_name = "ddb-table-%s" % short_uid()1137        kms_master_key_id = long_uid()1138        sse_specification = {"Enabled": True, "SSEType": "KMS", "KMSMasterKeyId": kms_master_key_id}1139        kms_master_key_arn = aws_stack.kms_key_arn(kms_master_key_id)1140        result = dynamodb_create_table_with_parameters(1141            TableName=table_name,1142            KeySchema=[{"AttributeName": PARTITION_KEY, "KeyType": "HASH"}],1143            AttributeDefinitions=[{"AttributeName": PARTITION_KEY, "AttributeType": "S"}],1144            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},1145            SSESpecification=sse_specification,1146            Tags=TEST_DDB_TAGS,1147        )1148        assert result["TableDescription"]["SSEDescription"]1149        assert result["TableDescription"]["SSEDescription"]["Status"] == "ENABLED"1150        assert result["TableDescription"]["SSEDescription"]["KMSMasterKeyArn"] == kms_master_key_arn1151    def test_dynamodb_create_table_with_partial_sse_specification(1152        self, dynamodb_create_table_with_parameters, kms_client1153    ):1154        table_name = "ddb-table-%s" % short_uid()1155        sse_specification = {"Enabled": True}1156        result = dynamodb_create_table_with_parameters(1157            TableName=table_name,1158            KeySchema=[{"AttributeName": PARTITION_KEY, "KeyType": "HASH"}],1159            AttributeDefinitions=[{"AttributeName": PARTITION_KEY, "AttributeType": "S"}],1160            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},1161            SSESpecification=sse_specification,1162            Tags=TEST_DDB_TAGS,1163        )1164        assert result["TableDescription"]["SSEDescription"]1165        assert result["TableDescription"]["SSEDescription"]["Status"] == "ENABLED"1166        assert result["TableDescription"]["SSEDescription"]["SSEType"] == "KMS"1167        assert "KMSMasterKeyArn" in result["TableDescription"]["SSEDescription"]1168        kms_master_key_arn = result["TableDescription"]["SSEDescription"]["KMSMasterKeyArn"]1169        result = kms_client.describe_key(KeyId=kms_master_key_arn)1170        assert result["KeyMetadata"]["KeyManager"] == "AWS"1171    def test_dynamodb_get_batch_items(self, dynamodb_client, dynamodb_create_table_with_parameters):1172        table_name = "ddb-table-%s" % short_uid()1173        dynamodb_create_table_with_parameters(1174            TableName=table_name,1175            KeySchema=[{"AttributeName": "PK", "KeyType": "HASH"}],1176            AttributeDefinitions=[{"AttributeName": "PK", "AttributeType": "S"}],1177            ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},1178        )1179        result = dynamodb_client.batch_get_item(1180            RequestItems={table_name: {"Keys": [{"PK": {"S": "test-key"}}]}}1181        )1182        assert list(result["Responses"])[0] == table_name1183    def test_dynamodb_streams_describe_with_exclusive_start_shard_id(1184        self, dynamodb_resource, dynamodb_create_table1185    ):1186        table_name = f"test-ddb-table-{short_uid()}"1187        ddbstreams = aws_stack.create_external_boto_client("dynamodbstreams")1188        dynamodb_create_table(1189            table_name=table_name,1190            partition_key=PARTITION_KEY,1191            stream_view_type="NEW_AND_OLD_IMAGES",1192        )1193        table = dynamodb_resource.Table(table_name)1194        response = ddbstreams.describe_stream(StreamArn=table.latest_stream_arn)1195        assert response["ResponseMetadata"]["HTTPStatusCode"] == 2001196        assert len(response["StreamDescription"]["Shards"]) == 11197        shard_id = response["StreamDescription"]["Shards"][0]["ShardId"]1198        response = ddbstreams.describe_stream(1199            StreamArn=table.latest_stream_arn, ExclusiveStartShardId=shard_id1200        )1201        assert response["ResponseMetadata"]["HTTPStatusCode"] == 2001202        assert len(response["StreamDescription"]["Shards"]) == 01203    @pytest.mark.aws_validated1204    def test_dynamodb_idempotent_writing(1205        self, dynamodb_create_table_with_parameters, dynamodb_client, dynamodb_wait_for_table_active1206    ):1207        table_name = f"ddb-table-{short_uid()}"1208        dynamodb_create_table_with_parameters(1209            TableName=table_name,1210            KeySchema=[1211                {"AttributeName": "id", "KeyType": "HASH"},1212                {"AttributeName": "name", "KeyType": "RANGE"},1213            ],1214            AttributeDefinitions=[1215                {"AttributeName": "id", "AttributeType": "S"},1216                {"AttributeName": "name", "AttributeType": "S"},1217            ],1218            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},1219        )1220        dynamodb_wait_for_table_active(table_name)1221        def _transact_write(_d: Dict):1222            response = dynamodb_client.transact_write_items(1223                ClientRequestToken="dedupe_token",1224                TransactItems=[1225                    {1226                        "Put": {1227                            "TableName": table_name,1228                            "Item": _d,1229                        }1230                    },1231                ],1232            )1233            assert response["ResponseMetadata"]["HTTPStatusCode"] == 2001234        _transact_write({"id": {"S": "id1"}, "name": {"S": "name1"}})1235        _transact_write({"name": {"S": "name1"}, "id": {"S": "id1"}})1236def delete_table(name):1237    dynamodb_client = aws_stack.create_external_boto_client("dynamodb")...

test_lambda_integration.py

Source:test_lambda_integration.py

1import base642import json3import os4import time5from unittest.mock import patch6import pytest7from botocore.exceptions import ClientError8from localstack import config9from localstack.services.apigateway.helpers import path_based_url10from localstack.services.awslambda.lambda_api import (11    BATCH_SIZE_RANGES,12    INVALID_PARAMETER_VALUE_EXCEPTION,13)14from localstack.services.awslambda.lambda_utils import LAMBDA_RUNTIME_PYTHON3615from localstack.utils import testutil16from localstack.utils.aws import aws_stack17from localstack.utils.common import retry, safe_requests, short_uid18from localstack.utils.sync import poll_condition19from localstack.utils.testutil import check_expected_lambda_log_events_length, get_lambda_log_events20from .test_lambda import (21    TEST_LAMBDA_FUNCTION_PREFIX,22    TEST_LAMBDA_LIBS,23    TEST_LAMBDA_PYTHON,24    TEST_LAMBDA_PYTHON_ECHO,25    is_old_provider,26)27TEST_STAGE_NAME = "testing"28TEST_SNS_TOPIC_NAME = "sns-topic-1"29THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))30TEST_LAMBDA_PARALLEL_FILE = os.path.join(THIS_FOLDER, "functions", "lambda_parallel.py")31class TestLambdaEventSourceMappings:32    def test_event_source_mapping_default_batch_size(33        self,34        create_lambda_function,35        lambda_client,36        sqs_client,37        sqs_create_queue,38        sqs_queue_arn,39        dynamodb_client,40        dynamodb_create_table,41        lambda_su_role,42    ):43        function_name = f"lambda_func-{short_uid()}"44        queue_name_1 = f"queue-{short_uid()}-1"45        queue_name_2 = f"queue-{short_uid()}-2"46        ddb_table = f"ddb_table-{short_uid()}"47        create_lambda_function(48            func_name=function_name,49            handler_file=TEST_LAMBDA_PYTHON_ECHO,50            runtime=LAMBDA_RUNTIME_PYTHON36,51            role=lambda_su_role,52        )53        queue_url_1 = sqs_create_queue(QueueName=queue_name_1)54        queue_arn_1 = sqs_queue_arn(queue_url_1)55        rs = lambda_client.create_event_source_mapping(56            EventSourceArn=queue_arn_1, FunctionName=function_name57        )58        assert BATCH_SIZE_RANGES["sqs"][0] == rs["BatchSize"]59        uuid = rs["UUID"]60        def wait_for_event_source_mapping():61            return lambda_client.get_event_source_mapping(UUID=uuid)["State"] == "Enabled"62        assert poll_condition(wait_for_event_source_mapping, timeout=30)63        with pytest.raises(ClientError) as e:64            # Update batch size with invalid value65            lambda_client.update_event_source_mapping(66                UUID=uuid,67                FunctionName=function_name,68                BatchSize=BATCH_SIZE_RANGES["sqs"][1] + 1,69            )70        e.match(INVALID_PARAMETER_VALUE_EXCEPTION)71        queue_url_2 = sqs_create_queue(QueueName=queue_name_2)72        queue_arn_2 = sqs_queue_arn(queue_url_2)73        with pytest.raises(ClientError) as e:74            # Create event source mapping with invalid batch size value75            lambda_client.create_event_source_mapping(76                EventSourceArn=queue_arn_2,77                FunctionName=function_name,78                BatchSize=BATCH_SIZE_RANGES["sqs"][1] + 1,79            )80        e.match(INVALID_PARAMETER_VALUE_EXCEPTION)81        table_description = dynamodb_create_table(82            table_name=ddb_table,83            partition_key="id",84            stream_view_type="NEW_IMAGE",85        )["TableDescription"]86        # table ARNs are not sufficient as event source, needs to be a dynamodb stream arn87        if not is_old_provider():88            with pytest.raises(ClientError) as e:89                lambda_client.create_event_source_mapping(90                    EventSourceArn=table_description["TableArn"],91                    FunctionName=function_name,92                    StartingPosition="LATEST",93                )94            e.match(INVALID_PARAMETER_VALUE_EXCEPTION)95        # check if event source mapping can be created with latest stream ARN96        rs = lambda_client.create_event_source_mapping(97            EventSourceArn=table_description["LatestStreamArn"],98            FunctionName=function_name,99            StartingPosition="LATEST",100        )101        assert BATCH_SIZE_RANGES["dynamodb"][0] == rs["BatchSize"]102    def test_disabled_event_source_mapping_with_dynamodb(103        self,104        create_lambda_function,105        lambda_client,106        dynamodb_resource,107        dynamodb_client,108        dynamodb_create_table,109        logs_client,110        dynamodbstreams_client,111        lambda_su_role,112    ):113        function_name = f"lambda_func-{short_uid()}"114        ddb_table = f"ddb_table-{short_uid()}"115        create_lambda_function(116            func_name=function_name,117            handler_file=TEST_LAMBDA_PYTHON_ECHO,118            runtime=LAMBDA_RUNTIME_PYTHON36,119            role=lambda_su_role,120        )121        latest_stream_arn = dynamodb_create_table(122            table_name=ddb_table, partition_key="id", stream_view_type="NEW_IMAGE"123        )["TableDescription"]["LatestStreamArn"]124        rs = lambda_client.create_event_source_mapping(125            FunctionName=function_name,126            EventSourceArn=latest_stream_arn,127            StartingPosition="TRIM_HORIZON",128            MaximumBatchingWindowInSeconds=1,129        )130        uuid = rs["UUID"]131        def wait_for_table_created():132            return (133                dynamodb_client.describe_table(TableName=ddb_table)["Table"]["TableStatus"]134                == "ACTIVE"135            )136        assert poll_condition(wait_for_table_created, timeout=30)137        def wait_for_stream_created():138            return (139                dynamodbstreams_client.describe_stream(StreamArn=latest_stream_arn)[140                    "StreamDescription"141                ]["StreamStatus"]142                == "ENABLED"143            )144        assert poll_condition(wait_for_stream_created, timeout=30)145        table = dynamodb_resource.Table(ddb_table)146        items = [147            {"id": short_uid(), "data": "data1"},148            {"id": short_uid(), "data": "data2"},149        ]150        table.put_item(Item=items[0])151        def assert_events():152            events = get_lambda_log_events(function_name, logs_client=logs_client)153            # lambda was invoked 1 time154            assert 1 == len(events[0]["Records"])155        # might take some time against AWS156        retry(assert_events, sleep=3, retries=10)157        # disable event source mapping158        lambda_client.update_event_source_mapping(UUID=uuid, Enabled=False)159        table.put_item(Item=items[1])160        events = get_lambda_log_events(function_name, logs_client=logs_client)161        # lambda no longer invoked, still have 1 event162        assert 1 == len(events[0]["Records"])163    # TODO invalid test against AWS, this behavior just is not correct164    def test_deletion_event_source_mapping_with_dynamodb(165        self, create_lambda_function, lambda_client, dynamodb_client, lambda_su_role166    ):167        function_name = f"lambda_func-{short_uid()}"168        ddb_table = f"ddb_table-{short_uid()}"169        create_lambda_function(170            func_name=function_name,171            handler_file=TEST_LAMBDA_PYTHON_ECHO,172            runtime=LAMBDA_RUNTIME_PYTHON36,173            role=lambda_su_role,174        )175        latest_stream_arn = aws_stack.create_dynamodb_table(176            table_name=ddb_table,177            partition_key="id",178            client=dynamodb_client,179            stream_view_type="NEW_IMAGE",180        )["TableDescription"]["LatestStreamArn"]181        lambda_client.create_event_source_mapping(182            FunctionName=function_name,183            EventSourceArn=latest_stream_arn,184            StartingPosition="TRIM_HORIZON",185        )186        def wait_for_table_created():187            return (188                dynamodb_client.describe_table(TableName=ddb_table)["Table"]["TableStatus"]189                == "ACTIVE"190            )191        assert poll_condition(wait_for_table_created, timeout=30)192        dynamodb_client.delete_table(TableName=ddb_table)193        result = lambda_client.list_event_source_mappings(EventSourceArn=latest_stream_arn)194        assert 1 == len(result["EventSourceMappings"])195    def test_event_source_mapping_with_sqs(196        self,197        create_lambda_function,198        lambda_client,199        sqs_client,200        sqs_create_queue,201        sqs_queue_arn,202        logs_client,203        lambda_su_role,204    ):205        function_name = f"lambda_func-{short_uid()}"206        queue_name_1 = f"queue-{short_uid()}-1"207        create_lambda_function(208            func_name=function_name,209            handler_file=TEST_LAMBDA_PYTHON_ECHO,210            runtime=LAMBDA_RUNTIME_PYTHON36,211            role=lambda_su_role,212        )213        queue_url_1 = sqs_create_queue(QueueName=queue_name_1)214        queue_arn_1 = sqs_queue_arn(queue_url_1)215        lambda_client.create_event_source_mapping(216            EventSourceArn=queue_arn_1, FunctionName=function_name, MaximumBatchingWindowInSeconds=1217        )218        sqs_client.send_message(QueueUrl=queue_url_1, MessageBody=json.dumps({"foo": "bar"}))219        def assert_lambda_log_events():220            events = get_lambda_log_events(function_name=function_name, logs_client=logs_client)221            # lambda was invoked 1 time222            assert 1 == len(events[0]["Records"])223        retry(assert_lambda_log_events, sleep_before=3, retries=30)224        rs = sqs_client.receive_message(QueueUrl=queue_url_1)225        assert rs.get("Messages") is None226    def test_create_kinesis_event_source_mapping(227        self,228        create_lambda_function,229        lambda_client,230        kinesis_client,231        kinesis_create_stream,232        lambda_su_role,233        wait_for_stream_ready,234        logs_client,235    ):236        function_name = f"lambda_func-{short_uid()}"237        stream_name = f"test-foobar-{short_uid()}"238        create_lambda_function(239            func_name=function_name,240            handler_file=TEST_LAMBDA_PYTHON_ECHO,241            runtime=LAMBDA_RUNTIME_PYTHON36,242            role=lambda_su_role,243        )244        kinesis_create_stream(StreamName=stream_name, ShardCount=1)245        stream_arn = kinesis_client.describe_stream(StreamName=stream_name)["StreamDescription"][246            "StreamARN"247        ]248        # only valid against AWS / new provider (once implemented)249        if not is_old_provider():250            with pytest.raises(ClientError) as e:251                lambda_client.create_event_source_mapping(252                    EventSourceArn=stream_arn, FunctionName=function_name253                )254            e.match(INVALID_PARAMETER_VALUE_EXCEPTION)255        wait_for_stream_ready(stream_name=stream_name)256        lambda_client.create_event_source_mapping(257            EventSourceArn=stream_arn, FunctionName=function_name, StartingPosition="TRIM_HORIZON"258        )259        stream_summary = kinesis_client.describe_stream_summary(StreamName=stream_name)260        assert 1 == stream_summary["StreamDescriptionSummary"]["OpenShardCount"]261        num_events_kinesis = 10262        kinesis_client.put_records(263            Records=[264                {"Data": "{}", "PartitionKey": f"test_{i}"} for i in range(0, num_events_kinesis)265            ],266            StreamName=stream_name,267        )268        def get_lambda_events():269            events = get_lambda_log_events(function_name, logs_client=logs_client)270            assert events271            return events272        events = retry(get_lambda_events, retries=30)273        assert 10 == len(events[0]["Records"])274        assert "eventID" in events[0]["Records"][0]275        assert "eventSourceARN" in events[0]["Records"][0]276        assert "eventSource" in events[0]["Records"][0]277        assert "eventVersion" in events[0]["Records"][0]278        assert "eventName" in events[0]["Records"][0]279        assert "invokeIdentityArn" in events[0]["Records"][0]280        assert "awsRegion" in events[0]["Records"][0]281        assert "kinesis" in events[0]["Records"][0]282    def test_python_lambda_subscribe_sns_topic(283        self,284        create_lambda_function,285        sns_client,286        lambda_su_role,287        sns_topic,288        logs_client,289        lambda_client,290    ):291        function_name = f"{TEST_LAMBDA_FUNCTION_PREFIX}-{short_uid()}"292        permission_id = f"test-statement-{short_uid()}"293        lambda_creation_response = create_lambda_function(294            func_name=function_name,295            handler_file=TEST_LAMBDA_PYTHON_ECHO,296            runtime=LAMBDA_RUNTIME_PYTHON36,297            role=lambda_su_role,298        )299        lambda_arn = lambda_creation_response["CreateFunctionResponse"]["FunctionArn"]300        topic_arn = sns_topic["Attributes"]["TopicArn"]301        lambda_client.add_permission(302            FunctionName=function_name,303            StatementId=permission_id,304            Action="lambda:InvokeFunction",305            Principal="sns.amazonaws.com",306            SourceArn=topic_arn,307        )308        sns_client.subscribe(309            TopicArn=topic_arn,310            Protocol="lambda",311            Endpoint=lambda_arn,312        )313        subject = "[Subject] Test subject"314        message = "Hello world."315        sns_client.publish(TopicArn=topic_arn, Subject=subject, Message=message)316        events = retry(317            check_expected_lambda_log_events_length,318            retries=10,319            sleep=1,320            function_name=function_name,321            expected_length=1,322            regex_filter="Records.*Sns",323            logs_client=logs_client,324        )325        notification = events[0]["Records"][0]["Sns"]326        assert "Subject" in notification327        assert subject == notification["Subject"]328class TestLambdaHttpInvocation:329    def test_http_invocation_with_apigw_proxy(self, create_lambda_function):330        lambda_name = f"test_lambda_{short_uid()}"331        lambda_resource = "/api/v1/{proxy+}"332        lambda_path = "/api/v1/hello/world"333        lambda_request_context_path = "/" + TEST_STAGE_NAME + lambda_path334        lambda_request_context_resource_path = lambda_resource335        # create lambda function336        create_lambda_function(337            func_name=lambda_name,338            handler_file=TEST_LAMBDA_PYTHON,339            libs=TEST_LAMBDA_LIBS,340        )341        # create API Gateway and connect it to the Lambda proxy backend342        lambda_uri = aws_stack.lambda_function_arn(lambda_name)343        target_uri = f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path/2015-03-31/functions/{lambda_uri}/invocations"344        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(345            "test_gateway2",346            target_uri,347            path=lambda_resource,348            stage_name=TEST_STAGE_NAME,349        )350        api_id = result["id"]351        url = path_based_url(api_id=api_id, stage_name=TEST_STAGE_NAME, path=lambda_path)352        result = safe_requests.post(353            url, data=b"{}", headers={"User-Agent": "python-requests/testing"}354        )355        content = json.loads(result.content)356        assert lambda_path == content["path"]357        assert lambda_resource == content["resource"]358        assert lambda_request_context_path == content["requestContext"]["path"]359        assert lambda_request_context_resource_path == content["requestContext"]["resourcePath"]360class TestKinesisSource:361    @patch.object(config, "SYNCHRONOUS_KINESIS_EVENTS", False)362    def test_kinesis_lambda_parallelism(363        self,364        lambda_client,365        kinesis_client,366        create_lambda_function,367        kinesis_create_stream,368        wait_for_stream_ready,369        logs_client,370        lambda_su_role,371    ):372        function_name = f"lambda_func-{short_uid()}"373        stream_name = f"test-foobar-{short_uid()}"374        create_lambda_function(375            handler_file=TEST_LAMBDA_PARALLEL_FILE,376            func_name=function_name,377            runtime=LAMBDA_RUNTIME_PYTHON36,378            role=lambda_su_role,379        )380        kinesis_create_stream(StreamName=stream_name, ShardCount=1)381        stream_arn = kinesis_client.describe_stream(StreamName=stream_name)["StreamDescription"][382            "StreamARN"383        ]384        wait_for_stream_ready(stream_name=stream_name)385        lambda_client.create_event_source_mapping(386            EventSourceArn=stream_arn,387            FunctionName=function_name,388            StartingPosition="TRIM_HORIZON",389            BatchSize=10,390        )391        stream_summary = kinesis_client.describe_stream_summary(StreamName=stream_name)392        assert 1 == stream_summary["StreamDescriptionSummary"]["OpenShardCount"]393        num_events_kinesis = 10394        # assure async call395        start = time.perf_counter()396        kinesis_client.put_records(397            Records=[398                {"Data": '{"batch": 0}', "PartitionKey": f"test_{i}"}399                for i in range(0, num_events_kinesis)400            ],401            StreamName=stream_name,402        )403        assert (time.perf_counter() - start) < 1  # this should not take more than a second404        kinesis_client.put_records(405            Records=[406                {"Data": '{"batch": 1}', "PartitionKey": f"test_{i}"}407                for i in range(0, num_events_kinesis)408            ],409            StreamName=stream_name,410        )411        def get_events():412            events = get_lambda_log_events(413                function_name, regex_filter=r"event.*Records", logs_client=logs_client414            )415            assert len(events) == 2416            return events417        events = retry(get_events, retries=30)418        def assertEvent(event, batch_no):419            assert 10 == len(event["event"]["Records"])420            assert "eventID" in event["event"]["Records"][0]421            assert "eventSourceARN" in event["event"]["Records"][0]422            assert "eventSource" in event["event"]["Records"][0]423            assert "eventVersion" in event["event"]["Records"][0]424            assert "eventName" in event["event"]["Records"][0]425            assert "invokeIdentityArn" in event["event"]["Records"][0]426            assert "awsRegion" in event["event"]["Records"][0]427            assert "kinesis" in event["event"]["Records"][0]428            assert {"batch": batch_no} == json.loads(429                base64.b64decode(event["event"]["Records"][0]["kinesis"]["data"]).decode(430                    config.DEFAULT_ENCODING431                )432            )433        assertEvent(events[0], 0)434        assertEvent(events[1], 1)...

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.