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.vpc_endpoints import ( 

28 _strip_deleted_vpc_endpoints, 

29) 

30 

31_BASELINE_EFS_ACCEPTANCE_STATE_KEY = "baseline_accepted_efs_automatic_backup_recovery_points" 

32 

33 

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

35 """Capture protected stacks/ECR and reject non-stack project leftovers.""" 

36 if ctx.checkpoint.baseline is not None: 

37 accepted_efs_backups = ctx.checkpoint.state.get( 

38 _BASELINE_EFS_ACCEPTANCE_STATE_KEY, 

39 [], 

40 ) 

41 if not isinstance(accepted_efs_backups, list): 

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

43 return { 

44 "reused_checkpoint_baseline": True, 

45 **ctx.checkpoint.baseline, 

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

47 } 

48 

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

50 if not enabled_regions: 

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

52 baseline = capture_baseline( 

53 ctx.session, 

54 enabled_regions=enabled_regions, 

55 ecr_regions=_topology_regions(ctx), 

56 protected_stack_names=ctx.settings.protected_stack_names, 

57 ) 

58 

59 project_inventory = collect_project_resources( 

60 ctx.session, 

61 enabled_regions=enabled_regions, 

62 expected_account=ctx.settings.expected_account, 

63 project_name=ctx.config.project_name, 

64 seed_region=ctx.config.global_region, 

65 validation_run_id=ctx.settings.run_id, 

66 ) 

67 disallowed_inventory = _strip_baseline_ecr(project_inventory, baseline) 

68 disallowed_inventory, accepted_efs_backups = ( 

69 _strip_accepted_efs_automatic_backup_recovery_points( 

70 ctx, 

71 disallowed_inventory, 

72 ) 

73 ) 

74 disallowed_inventory, accepted_expired_streams = _strip_expired_table_streams( 

75 ctx, 

76 disallowed_inventory, 

77 ) 

78 disallowed_inventory, accepted_deleted_vpc_endpoints = _strip_deleted_vpc_endpoints( 

79 ctx, 

80 disallowed_inventory, 

81 ) 

82 if not project_resources_are_absent(disallowed_inventory): 

83 raise RuntimeError( 

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

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

86 ) 

87 

88 ctx.checkpoint.baseline = baseline 

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

90 ctx.persist() 

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

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

93 # capture that final-inventory's compare_baseline expects. 

94 return { 

95 **baseline, 

96 "accepted_efs_automatic_backup_recovery_points": accepted_efs_backups, 

97 "accepted_expired_dynamodb_streams": accepted_expired_streams, 

98 "accepted_deleted_vpc_endpoints": accepted_deleted_vpc_endpoints, 

99 }