Create a table with streams enabled, yield (table_name, stream_arn), cleanup.
(dynamodb_client)
| 50 | return boto3.client(**kwargs) |
| 51 | @pytest.fixture() |
| 52 | def stream_table(dynamodb_client): |
| 53 | """Create a table with streams enabled, yield (table_name, stream_arn), cleanup.""" |
| 54 | table_name = f"extenddb-stream-{uuid.uuid4().hex[:8]}" |
| 55 | resp = dynamodb_client.create_table( |
| 56 | TableName=table_name, |
| 57 | AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], |
| 58 | KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], |
| 59 | BillingMode="PAY_PER_REQUEST", |
| 60 | StreamSpecification={ |
| 61 | "StreamEnabled": True, |
| 62 | "StreamViewType": "NEW_AND_OLD_IMAGES", |
| 63 | }, |
| 64 | ) |
| 65 | wait_for_active(dynamodb_client, table_name) |
| 66 | desc = dynamodb_client.describe_table(TableName=table_name) |
| 67 | stream_arn = desc["Table"]["LatestStreamArn"] |
| 68 | yield table_name, stream_arn |
| 69 | try: |
| 70 | dynamodb_client.delete_table(TableName=table_name) |
| 71 | wait_for_deleted(dynamodb_client, table_name) |
| 72 | except Exception: |
| 73 | pass |
| 74 | def _drain_all_shards( |
| 75 | streams_client, stream_arn: str, iterator_type: str = "TRIM_HORIZON", |
| 76 | sequence_number: str | None = None, max_polls: int = 20, |
nothing calls this directly
no test coverage detected