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

1"""baseline: capture protected CloudFormation and ECR baselines.""" 

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 project_resources_are_absent, 

16) 

17from ..models import RunContext 

18from ..ownership.dynamodb_streams import ( 

19 _strip_expired_table_streams, 

20) 

21from ..ownership.ecr import ( 

22 _strip_baseline_ecr, 

23) 

24from ..ownership.efs_automatic_backups import ( 

25 _strip_accepted_efs_automatic_backup_recovery_points, 

26) 

27from ..ownership.transaction_search import ( 

28 TRANSACTION_SEARCH_BASELINE_STATE_KEY, 

29 _capture_transaction_search_baseline, 

30 _validated_transaction_search_baseline, 

31) 

32from ..ownership.vpc_endpoints import ( 

33 _strip_deleted_vpc_endpoints, 

34) 

35 

36_BASELINE_EFS_ACCEPTANCE_STATE_KEY = "baseline_accepted_efs_automatic_backup_recovery_points" 

37 

38 

39def action_baseline(ctx: RunContext) -> dict[str, Any]: 

40 """Capture protected stacks/ECR and Transaction Search state; reject project leftovers. 

41 

42 The Transaction Search record (every regional Region's X-Ray trace segment 

43 destination, GCO resource policy, and span log groups) is persisted with 

44 the baseline in the same checkpoint write, because teardown restores 

45 exactly what it says and a resume must never re-capture state the run 

46 already changed. 

47 """ 

48 if ctx.checkpoint.baseline is not None: 

49 accepted_efs_backups = ctx.checkpoint.state.get( 

50 _BASELINE_EFS_ACCEPTANCE_STATE_KEY, 

51 [], 

52 ) 

53 if not isinstance(accepted_efs_backups, list): 

54 raise RuntimeError("Checkpoint baseline EFS acceptance evidence must be a list") 

55 transaction_search = _validated_transaction_search_baseline(ctx) 

56 return { 

57 "reused_checkpoint_baseline": True, 

58 **ctx.checkpoint.baseline, 

59 "accepted_efs_automatic_backup_recovery_points": copy.deepcopy(accepted_efs_backups), 

60 "transaction_search": copy.deepcopy(transaction_search), 

61 } 

62 

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

64 if not enabled_regions: 

65 raise RuntimeError("Preflight did not record enabled AWS Regions") 

66 baseline = capture_baseline( 

67 ctx.session, 

68 enabled_regions=enabled_regions, 

69 ecr_regions=_topology_regions(ctx), 

70 protected_stack_names=ctx.settings.protected_stack_names, 

71 ) 

72 

73 project_inventory = collect_project_resources( 

74 ctx.session, 

75 enabled_regions=enabled_regions, 

76 expected_account=ctx.settings.expected_account, 

77 project_name=ctx.config.project_name, 

78 seed_region=ctx.config.global_region, 

79 validation_run_id=ctx.settings.run_id, 

80 ) 

81 disallowed_inventory = _strip_baseline_ecr(project_inventory, baseline) 

82 disallowed_inventory, accepted_efs_backups = ( 

83 _strip_accepted_efs_automatic_backup_recovery_points( 

84 ctx, 

85 disallowed_inventory, 

86 ) 

87 ) 

88 disallowed_inventory, accepted_expired_streams = _strip_expired_table_streams( 

89 ctx, 

90 disallowed_inventory, 

91 ) 

92 disallowed_inventory, accepted_deleted_vpc_endpoints = _strip_deleted_vpc_endpoints( 

93 ctx, 

94 disallowed_inventory, 

95 ) 

96 if not project_resources_are_absent(disallowed_inventory): 

97 raise RuntimeError( 

98 "Fresh baseline contains project resources not owned by this run: " 

99 + json.dumps(disallowed_inventory, sort_keys=True) 

100 ) 

101 

102 transaction_search = _capture_transaction_search_baseline(ctx) 

103 

104 ctx.checkpoint.baseline = baseline 

105 ctx.checkpoint.state[_BASELINE_EFS_ACCEPTANCE_STATE_KEY] = copy.deepcopy(accepted_efs_backups) 

106 ctx.checkpoint.state[TRANSACTION_SEARCH_BASELINE_STATE_KEY] = copy.deepcopy(transaction_search) 

107 ctx.persist() 

108 # The accepted-stream evidence rides the action result (report) only; the 

109 # persisted checkpoint baseline stays exactly the protected-stack/ECR 

110 # capture that final-inventory's compare_baseline expects. The Transaction 

111 # Search record lives in checkpoint state for the same reason. 

112 return { 

113 **baseline, 

114 "accepted_efs_automatic_backup_recovery_points": accepted_efs_backups, 

115 "accepted_expired_dynamodb_streams": accepted_expired_streams, 

116 "accepted_deleted_vpc_endpoints": accepted_deleted_vpc_endpoints, 

117 "transaction_search": transaction_search, 

118 }