Coverage for gco_mcp / resources / _eks.py: 100.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_MAX_REGION_LENGTH = 32 

16 

17 

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

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

20 return len(region) <= _MAX_REGION_LENGTH and _REGION_RE.fullmatch(region) is not None 

21 

22 

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

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

25 

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

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

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

29 """ 

30 if not is_valid_region(region): 

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

32 

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

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

35 raise ValueError("invalid GCO project name") 

36 

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

38 if _ACCOUNT_ID_RE.fullmatch(account) is None: 

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

40 

41 session = boto3.session.Session() 

42 partition = session.get_partition_for_region(region) 

43 if not isinstance(partition, str) or _PARTITION_RE.fullmatch(partition) is None: 

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

45 

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

47 return f"arn:{partition}:eks:{region}:{account}:cluster/{cluster_name}"