Coverage for scripts / live_release_validation / cleanup / ecr.py: 100.00%

1"""Delete exactly run-created ECR images and repositories.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ..constants import ( 

8 _RUN_STACK_TAG, 

9) 

10from ..inventory import ( 

11 collect_ecr_inventory, 

12 describe_ecr_image_by_tag, 

13) 

14from ..models import RunContext 

15from ..ownership.ecr import ( 

16 _ecr_creation_identity, 

17 _ecr_image_identity, 

18) 

19 

20 

21def _cleanup_new_ecr_images(ctx: RunContext) -> dict[str, Any]: 

22 """Retain mutable baseline-repository tag deltas that cannot be conditionally deleted.""" 

23 retained: list[dict[str, Any]] = [] 

24 for record in ctx.checkpoint.state.get("retained_ecr_image_deltas", []): 

25 reference = f"{record['region']}:{record['repository']}:{record['tag']}" 

26 current = describe_ecr_image_by_tag( 

27 ctx.session, 

28 region=str(record["region"]), 

29 repository_name=str(record["repository"]), 

30 tag=str(record["tag"]), 

31 ) 

32 if current is None: 

33 retained.append({"image": reference, "already_absent": True}) 

34 continue 

35 if _ecr_image_identity(current) != record.get("identity"): 

36 raise RuntimeError(f"Observed ECR image identity changed for {reference}") 

37 retained.append( 

38 { 

39 "image": reference, 

40 "digest": current["digest"], 

41 "retained": True, 

42 "reason": "ECR has no conditional tag deletion primitive", 

43 } 

44 ) 

45 return {"images": retained, "automatic_deletion": False} 

46 

47 

48def _cleanup_new_ecr_repositories(ctx: RunContext) -> dict[str, Any]: 

49 """Retain acknowledged repositories because ECR deletion is not conditional.""" 

50 results: list[dict[str, Any]] = [] 

51 records = ctx.checkpoint.state.get("created_ecr_repositories", []) 

52 current_by_region = ( 

53 collect_ecr_inventory(ctx.session, {str(item["region"]) for item in records}) 

54 if records 

55 else {} 

56 ) 

57 for record in records: 

58 region = str(record["region"]) 

59 current = next( 

60 ( 

61 item 

62 for item in current_by_region.get(region, []) 

63 if item.get("name") == record["name"] 

64 ), 

65 None, 

66 ) 

67 if current is None: 

68 results.append({"arn": record["arn"], "already_absent": True}) 

69 continue 

70 if _ecr_creation_identity(current) != record.get("creation_identity"): 

71 raise RuntimeError( 

72 f"ECR repository creation identity changed for {region}:{record['name']}" 

73 ) 

74 if (current.get("tags") or {}).get(_RUN_STACK_TAG) != record.get("run_tag"): 

75 raise RuntimeError(f"ECR run ownership changed for {record['arn']}") 

76 results.append( 

77 { 

78 "arn": record["arn"], 

79 "retained": True, 

80 "reason": "ECR has no conditional repository deletion primitive", 

81 } 

82 ) 

83 return {"repositories": results, "automatic_deletion": False}