Coverage for gco_mcp/resources/_eks.py: 85.00%

1"""Explicit EKS context resolution shared by live MCP resources.""" 

2 

3from __future__ import annotations 

4 

5import re 

6 

7import boto3 

8 

9from cli.config import get_config 

10 

11_REGION_RE = re.compile(r"^[a-z]{2,4}(?:-[a-z0-9]+)+-[0-9]+$") 

12_PARTITION_RE = re.compile(r"^[a-z][a-z0-9-]*$") 

13_PROJECT_NAME_RE = re.compile(r"^[a-z][a-z0-9-]{1,30}$") 

14_ACCOUNT_ID_RE = re.compile(r"^[0-9]{12}$") 

15_EKS_ARN_RE = re.compile( 

16 r"^arn:([a-z][a-z0-9-]*):eks:([a-z0-9-]+):([0-9]{12}):" 

17 r"cluster/([a-z][a-z0-9-]{1,99})$" 

18) 

19 

20 

21def is_valid_region(region: str) -> bool: 

22 """Return whether ``region`` has the bounded AWS region shape GCO accepts.""" 

23 return _REGION_RE.fullmatch(region) is not None 

24 

25 

26def eks_context_for_region(region: str, project_name: str | None = None) -> str: 

27 """Return an account-qualified kubectl context ARN for one GCO EKS cluster. 

28 

29 The cluster prefix follows the same merged CLI configuration as every other 

30 project-scoped command (cdk.json, config file, then ``GCO_PROJECT_NAME``). 

31 An explicit ``project_name`` is accepted for deterministic callers/tests. 

32 """ 

33 if not is_valid_region(region): 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 raise ValueError(f"invalid AWS region: {region}") 

35 

36 project = project_name if project_name is not None else get_config().project_name 

37 if not isinstance(project, str) or _PROJECT_NAME_RE.fullmatch(project) is None: 

38 raise ValueError("invalid GCO project name") 

39 

40 account = str(boto3.client("sts", region_name=region).get_caller_identity().get("Account", "")) 

41 if _ACCOUNT_ID_RE.fullmatch(account) is None: 

42 raise ValueError("STS returned an invalid AWS account ID") 

43 

44 session = boto3.session.Session() 

45 partition = session.get_partition_for_region(region) 

46 if not isinstance(partition, str) or _PARTITION_RE.fullmatch(partition) is None: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 raise ValueError(f"AWS SDK returned an invalid partition for region {region}") 

48 

49 cluster_name = f"{project}-{region}" 

50 arn = f"arn:{partition}:eks:{region}:{account}:cluster/{cluster_name}" 

51 match = _EKS_ARN_RE.fullmatch(arn) 

52 if ( 52 ↛ 58line 52 didn't jump to line 58 because the condition on line 52 was never true

53 match is None 

54 or match.group(1) != partition 

55 or match.group(2) != region 

56 or match.group(4) != cluster_name 

57 ): 

58 raise ValueError("failed to construct a valid EKS cluster ARN") 

59 return arn