Method test_insert_modify_remove_events

tests/test_streams.py:577–630

(self, dynamodb_client, streams_client, stream_table)

    """Inserts, updates, deletes generating INSERT/MODIFY/REMOVE events."""

    def test_insert_modify_remove_events(self, dynamodb_client, streams_client, stream_table):
        table_name, stream_arn = stream_table
        pk = f"mixed-{uuid.uuid4().hex[:8]}"

        # INSERT
        dynamodb_client.put_item(
            TableName=table_name,
            Item={"pk": {"S": pk}, "val": {"N": "1"}},
        )
        # MODIFY
        dynamodb_client.update_item(
            TableName=table_name,
            Key={"pk": {"S": pk}},
            UpdateExpression="SET val = :v",
            ExpressionAttributeValues={":v": {"N": "2"}},
        )
        # REMOVE
        dynamodb_client.delete_item(
            TableName=table_name,
            Key={"pk": {"S": pk}},
        )

        # Poll until we have all 3 events for this pk.
        deadline = time.monotonic() + 5.0
        our_records: list[dict] = []
        while time.monotonic() < deadline:
            records = _drain_all_shards(streams_client, stream_arn, "TRIM_HORIZON")
            our_records = [
                r for r in records
                if r["dynamodb"]["Keys"]["pk"]["S"] == pk
            ]
            if len(our_records) >= 3:
                break
            time.sleep(_poll_interval())

        events = [r["eventName"] for r in our_records]
        assert events == ["INSERT", "MODIFY", "REMOVE"], (
            f"Expected [INSERT, MODIFY, REMOVE], got {events}"
        )

        # Verify images.
        insert_rec = our_records[0]
        assert "NewImage" in insert_rec["dynamodb"]
        assert insert_rec["dynamodb"]["NewImage"]["val"]["N"] == "1"

        modify_rec = our_records[1]
        assert "OldImage" in modify_rec["dynamodb"]
        assert modify_rec["dynamodb"]["OldImage"]["val"]["N"] == "1"
        assert "NewImage" in modify_rec["dynamodb"]
        assert modify_rec["dynamodb"]["NewImage"]["val"]["N"] == "2"

        remove_rec = our_records[2]
        assert "OldImage" in remove_rec["dynamodb"]
        assert remove_rec["dynamodb"]["OldImage"]["val"]["N"] == "2"
class TestGetShardIterator:
    """Edge cases for GetShardIterator."""