Coverage for cli / kubectl_helpers.py: 87%

1""" 

2Shared kubectl helper utilities for GCO CLI. 

3 

4Provides common kubectl operations used across multiple CLI modules 

5to reduce code duplication and ensure consistent error handling. 

6""" 

7 

8import logging 

9import re 

10import subprocess 

11 

12logger = logging.getLogger(__name__) 

13 

14# EKS cluster names: 1-100 chars, alphanumeric and hyphens only. 

15# AWS region names: e.g. us-east-1, ap-southeast-2, eu-central-1. 

16_CLUSTER_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9\-]{0,99}$") 

17_REGION_RE = re.compile(r"^[a-z]{2,3}-[a-z]+-\d+$") 

18 

19 

20def _validate_cluster_name(cluster_name: str) -> None: 

21 """Raise ValueError if cluster_name contains characters outside the EKS naming rules.""" 

22 if not _CLUSTER_NAME_RE.match(cluster_name): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 raise ValueError( 

24 f"Invalid cluster name {cluster_name!r}: must be 1-100 alphanumeric/hyphen characters" 

25 ) 

26 

27 

28def _validate_region(region: str) -> None: 

29 """Raise ValueError if region does not match the standard AWS region pattern.""" 

30 if not _REGION_RE.match(region): 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 raise ValueError(f"Invalid AWS region {region!r}: expected format like 'us-east-1'") 

32 

33 

34def update_kubeconfig(cluster_name: str, region: str) -> None: 

35 """Update kubeconfig for an EKS cluster. 

36 

37 Args: 

38 cluster_name: Name of the EKS cluster 

39 region: AWS region where the cluster is located 

40 

41 Raises: 

42 ValueError: If cluster_name or region contain unexpected characters 

43 RuntimeError: If the kubeconfig update fails 

44 FileNotFoundError: If the AWS CLI is not installed 

45 """ 

46 _validate_cluster_name(cluster_name) 

47 _validate_region(region) 

48 

49 cmd = [ 

50 "aws", 

51 "eks", 

52 "update-kubeconfig", 

53 "--name", 

54 cluster_name, 

55 "--region", 

56 region, 

57 ] 

58 

59 try: 

60 result = subprocess.run( 

61 cmd, capture_output=True, text=True 

62 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True 

63 if result.returncode != 0: 

64 raise RuntimeError(f"Failed to update kubeconfig: {result.stderr}") 

65 except subprocess.CalledProcessError as e: 

66 raise RuntimeError(f"Failed to update kubeconfig: {e.stderr}") from e 

67 except FileNotFoundError as e: 

68 raise RuntimeError( 

69 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH." 

70 ) from e