Coverage for scripts / live_release_validation / ownership / dynamodb_streams.py: 100.00%

1"""Accepted-residue accounting for DynamoDB streams of deleted tables. 

2 

3Deleting a DynamoDB table does not delete its stream: the stream object stays 

4readable (``DISABLED``) for roughly 24 hours so consumers can drain it, there 

5is no delete API for it, and the Resource Groups Tagging API keeps returning 

6its ARN for as long as the index remembers it — with or without tags. The 

7project scanners flag such an ARN by name alone (``.../table/gco-.../stream/...``), 

8so a run that correctly destroyed its vector-store table still presents a 

9"project resource" that nothing can remove. Observed live on 2026-08-20: the 

10previous run's two ``gco-vector-store`` stream ARNs failed the next run's 

11``baseline`` gate 4.5 hours after their tables were destroyed, with the 

12streams still describable and ``DISABLED``. 

13 

14``_strip_expired_table_streams`` accepts exactly that shape and nothing else: 

15a ``tagged_resources`` entry whose ARN parses as a table stream in the same 

16region and expected account is stripped **only after** DynamoDB itself 

17confirms the parent table does not exist. A live table keeps its stream entry 

18in the inventory (genuine residue — and the table scanner reports the table 

19itself too). Every accepted entry is returned as evidence (region, table, 

20stream status, the check performed) so ``baseline`` and ``final-inventory`` 

21disclose what they tolerated rather than silently ignoring it, mirroring the 

22pending-deletion KMS precedent in ``ownership/kms.py``. 

23""" 

24 

25from __future__ import annotations 

26 

27import copy 

28import re 

29from typing import Any 

30 

31from botocore.exceptions import ClientError 

32 

33from ..models import RunContext 

34 

35_TABLE_STREAM_ARN = re.compile( 

36 r"^arn:[^:]+:dynamodb:(?P<region>[a-z0-9-]+):(?P<account>\d{12})" 

37 r":table/(?P<table>[^/]+)/stream/(?P<label>.+)$" 

38) 

39 

40 

41def _stream_status(ctx: RunContext, region: str, stream_arn: str) -> str: 

42 """Return the live stream status, or ``ABSENT`` once fully expired.""" 

43 streams = ctx.session.client("dynamodbstreams", region_name=region) 

44 try: 

45 description = streams.describe_stream(StreamArn=stream_arn).get("StreamDescription", {}) 

46 except ClientError as exc: 

47 if exc.response.get("Error", {}).get("Code") != "ResourceNotFoundException": 

48 raise 

49 return "ABSENT" 

50 return str(description.get("StreamStatus") or "UNKNOWN") 

51 

52 

53def _strip_expired_table_streams( 

54 ctx: RunContext, 

55 project_inventory: dict[str, Any], 

56) -> tuple[dict[str, Any], list[dict[str, Any]]]: 

57 """Strip tagged stream ARNs whose parent table DynamoDB proves absent.""" 

58 inventory = copy.deepcopy(project_inventory) 

59 accepted: list[dict[str, Any]] = [] 

60 for region, resources in list(inventory.get("regional", {}).items()): 

61 kept: list[dict[str, Any]] = [] 

62 for entry in resources.get("tagged_resources", []): 

63 arn = str(entry.get("arn") or "") 

64 match = _TABLE_STREAM_ARN.match(arn) 

65 if ( 

66 match is None 

67 or match.group("region") != region 

68 or match.group("account") != ctx.settings.expected_account 

69 ): 

70 kept.append(entry) 

71 continue 

72 table_name = match.group("table") 

73 dynamodb = ctx.session.client("dynamodb", region_name=region) 

74 try: 

75 dynamodb.describe_table(TableName=table_name) 

76 except ClientError as exc: 

77 if exc.response.get("Error", {}).get("Code") != "ResourceNotFoundException": 

78 raise 

79 else: 

80 # The table is live, so the stream is real project residue — 

81 # keep it (the table scanner reports the table itself too). 

82 kept.append(entry) 

83 continue 

84 accepted.append( 

85 { 

86 "region": region, 

87 "arn": arn, 

88 "table_name": table_name, 

89 "table_absent": True, 

90 "authority": "dynamodb:DescribeTable ResourceNotFoundException", 

91 "stream_status": _stream_status(ctx, region, arn), 

92 "tags": dict(entry.get("tags") or {}), 

93 "note": ( 

94 "streams of deleted tables have no delete API and " 

95 "expire on their own roughly 24h after table deletion" 

96 ), 

97 } 

98 ) 

99 resources["tagged_resources"] = kept 

100 if not any(resources.values()): 

101 inventory["regional"].pop(region) 

102 return inventory, accepted