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

1"""eks-capabilities: the opt-in AWS-managed ACK / kro are attached and ACTIVE.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ..checks.eks_capabilities import ( 

8 effective_eks_capabilities_config, 

9 enabled_types_by_region, 

10 verify_capabilities_attached, 

11) 

12from ..models import RunContext 

13 

14 

15def action_eks_capabilities(ctx: RunContext) -> dict[str, Any]: 

16 """Require every configured EKS Capability to be attached, ACTIVE and drift-free. 

17 

18 Resolves the block the run deployed with (cdk.json ``eks_capabilities`` 

19 merged with the run's ``--eks-capabilities`` selection). When no type is 

20 enabled for any deployed Region the action passes with a note: the shipped 

21 default is every capability off, and that default is what most runs 

22 validate. Otherwise, for every Region with an enabled type, the EKS API 

23 must report each enabled capability attached to the Region's cluster and 

24 ``ACTIVE`` with no drift from the configuration, judged by the same merge 

25 ``gco stacks capabilities status`` prints. AWS API only — no cluster 

26 session is needed. 

27 """ 

28 config = effective_eks_capabilities_config(ctx) 

29 by_region = enabled_types_by_region(config, ctx.deployment_regions) 

30 if not by_region: 

31 evidence: dict[str, Any] = { 

32 "enabled": False, 

33 "detail": ( 

34 "No EKS capability is enabled for any deployed Region (the shipped default); " 

35 "pass --eks-capabilities ack,kro (or all) to prove them" 

36 ), 

37 } 

38 ctx.checkpoint.state["eks_capabilities_validation"] = evidence 

39 ctx.persist() 

40 return evidence 

41 

42 regions: dict[str, Any] = {} 

43 for region, enabled_types in by_region.items(): 

44 record: dict[str, Any] = {"enabled_types": enabled_types} 

45 regions[region] = record 

46 ctx.checkpoint.state["eks_capabilities_validation"] = {"enabled": True, "regions": regions} 

47 ctx.persist() 

48 

49 status = verify_capabilities_attached(ctx, region, config) 

50 record["cluster_arn"] = status["cluster_arn"] 

51 record["capabilities"] = [ 

52 { 

53 key: row.get(key) 

54 for key in ("type", "capability_name", "status", "version", "arn", "role_arn") 

55 } 

56 for row in status["capabilities"] 

57 if row["configured"] 

58 ] 

59 ctx.persist() 

60 

61 evidence = {"enabled": True, "regions": regions} 

62 ctx.checkpoint.state["eks_capabilities_validation"] = evidence 

63 ctx.persist() 

64 return evidence