Coverage for scripts / live_release_validation / actions / final_inventory.py: 100.00%

1"""final-inventory: verify zero residual resources and baseline preservation.""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import json 

7from typing import Any 

8 

9from ..context import ( 

10 _topology_regions, 

11) 

12from ..inventory import ( 

13 capture_baseline, 

14 collect_project_resources, 

15 compare_baseline, 

16 project_resources_are_absent, 

17 summarize_project_resources, 

18) 

19from ..models import RunContext, utc_now 

20from ..ownership.dynamodb_streams import ( 

21 _strip_expired_table_streams, 

22) 

23from ..ownership.ecr import ( 

24 _strip_accepted_retained_ecr, 

25 _strip_baseline_ecr, 

26 _strip_expected_retained_ecr, 

27) 

28from ..ownership.efs_automatic_backups import ( 

29 _strip_accepted_efs_automatic_backup_recovery_points, 

30) 

31from ..ownership.kms import ( 

32 _strip_expected_pending_kms, 

33) 

34from ..ownership.stacks import ( 

35 _verify_target_stack_absence, 

36) 

37from ..ownership.vpc_endpoints import ( 

38 _strip_deleted_vpc_endpoints, 

39) 

40 

41 

42def action_final_inventory(ctx: RunContext) -> dict[str, Any]: 

43 """Prove cleanup and exact protected-stack/ECR baseline preservation.""" 

44 if ctx.checkpoint.baseline is None: 

45 raise RuntimeError("Final inventory cannot compare without a baseline") 

46 enabled_regions = ctx.checkpoint.state.get("enabled_regions") 

47 if not enabled_regions: 

48 raise RuntimeError("Checkpoint omitted enabled Regions") 

49 

50 stack_absence = _verify_target_stack_absence(ctx) 

51 final_baseline = capture_baseline( 

52 ctx.session, 

53 enabled_regions=enabled_regions, 

54 ecr_regions=ctx.checkpoint.baseline.get("ecr_regions") or _topology_regions(ctx), 

55 protected_stack_names=ctx.settings.protected_stack_names, 

56 ) 

57 comparison_baseline, accepted_retained_ecr = _strip_expected_retained_ecr( 

58 ctx, 

59 final_baseline, 

60 ) 

61 differences = compare_baseline(ctx.checkpoint.baseline, comparison_baseline) 

62 project_inventory = collect_project_resources( 

63 ctx.session, 

64 enabled_regions=enabled_regions, 

65 expected_account=ctx.settings.expected_account, 

66 project_name=ctx.config.project_name, 

67 seed_region=ctx.config.global_region, 

68 validation_run_id=ctx.settings.run_id, 

69 ) 

70 residual_inventory = _strip_baseline_ecr( 

71 project_inventory, 

72 ctx.checkpoint.baseline, 

73 ) 

74 residual_inventory = _strip_accepted_retained_ecr( 

75 residual_inventory, 

76 accepted_retained_ecr, 

77 ) 

78 residual_inventory, accepted_pending_kms = _strip_expected_pending_kms( 

79 ctx, 

80 residual_inventory, 

81 ) 

82 residual_inventory, accepted_efs_backups = _strip_accepted_efs_automatic_backup_recovery_points( 

83 ctx, residual_inventory 

84 ) 

85 residual_inventory, accepted_expired_streams = _strip_expired_table_streams( 

86 ctx, 

87 residual_inventory, 

88 ) 

89 residual_inventory, accepted_deleted_vpc_endpoints = _strip_deleted_vpc_endpoints( 

90 ctx, 

91 residual_inventory, 

92 ) 

93 summary = summarize_project_resources(residual_inventory) 

94 result = { 

95 "summary": summary, 

96 "stack_absence": stack_absence, 

97 "baseline_differences": differences, 

98 "protected_and_ecr_inventory": final_baseline, 

99 "comparison_inventory": comparison_baseline, 

100 "accepted_retained_ecr": accepted_retained_ecr, 

101 "project_resources": project_inventory, 

102 "accepted_pending_kms_keys": accepted_pending_kms, 

103 "accepted_efs_automatic_backup_recovery_points": accepted_efs_backups, 

104 "accepted_expired_dynamodb_streams": accepted_expired_streams, 

105 "accepted_deleted_vpc_endpoints": accepted_deleted_vpc_endpoints, 

106 "residual_project_resources": residual_inventory, 

107 } 

108 ctx.report.final_inventory = result 

109 ctx.checkpoint.state["final_inventory"] = copy.deepcopy(result) 

110 if not stack_absence["all_absent"] and ctx.checkpoint.destroyed: 

111 ctx.checkpoint.destroyed = False 

112 for action_name in ("destroy", "final-inventory"): 

113 if action_name in ctx.checkpoint.completed_actions: 

114 ctx.checkpoint.completed_actions.remove(action_name) 

115 ctx.checkpoint.state.setdefault("stale_destroyed_reconciliations", []).append( 

116 {"at": utc_now(), "stack_absence": stack_absence, "source": "final-inventory"} 

117 ) 

118 ctx.persist() 

119 if not stack_absence["all_absent"]: 

120 raise RuntimeError( 

121 "Target stacks remain after teardown: " 

122 + json.dumps(stack_absence["residual"], sort_keys=True) 

123 ) 

124 if differences: 

125 raise RuntimeError( 

126 "Protected stack/ECR baseline changed: " + json.dumps(differences, sort_keys=True) 

127 ) 

128 if not project_resources_are_absent(residual_inventory): 

129 raise RuntimeError( 

130 "Project resources remain after teardown: " 

131 + json.dumps(residual_inventory, sort_keys=True) 

132 ) 

133 return result