Coverage for gco_mcp/resources/jobs.py: 89.74%

1"""Live job state resources for explicitly selected regional clusters.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import re 

7from typing import Any 

8 

9import cli_runner 

10 

11from resources._eks import eks_context_for_region, is_valid_region 

12 

13# RFC 1123 label format. Job names live in the same namespace as pod 

14# names, so the same rule applies. Bounded length stops accidental 

15# command-line stuffing through a malformed URI template expansion. 

16_JOB_NAME_RE = re.compile(r"^[a-z0-9](?:[-a-z0-9]{0,251}[a-z0-9])?$") 

17_DEFAULT_NAMESPACE = "gco-jobs" 

18_KUBECTL_TIMEOUT_SECONDS = 30 

19 

20 

21def _job_resource(job_name: str) -> str: 

22 """Fail closed for the legacy URI that did not identify a cluster.""" 

23 return json.dumps( 

24 { 

25 "error": "explicit region required", 

26 "code": "eks_region_required", 

27 "use": f"gco://jobs/{{region}}/{job_name}", 

28 } 

29 ) 

30 

31 

32def _job_resource_for_region(region: str, job_name: str) -> str: 

33 """Return live YAML for ``job_name`` from one explicit regional cluster.""" 

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

35 return json.dumps({"error": "invalid region", "value": region}) 

36 if not _JOB_NAME_RE.fullmatch(job_name): 

37 return json.dumps( 

38 { 

39 "error": "invalid job_name", 

40 "detail": "must match ^[a-z0-9](?:[-a-z0-9]{0,251}[a-z0-9])?$", 

41 "value": job_name, 

42 } 

43 ) 

44 try: 

45 context_arn = eks_context_for_region(region) 

46 except Exception as exc: # AWS credential/session failures become resource errors 

47 return json.dumps({"error": "unable to resolve EKS context", "detail": str(exc)[:200]}) 

48 try: 

49 result = cli_runner.subprocess.run( # type: ignore[attr-defined] # nosemgrep: dangerous-subprocess-use-audit - shell=False; caller input is validated and the context ARN is constructed internally 

50 [ 

51 "kubectl", 

52 "get", 

53 "job", 

54 job_name, 

55 "-n", 

56 _DEFAULT_NAMESPACE, 

57 "-o", 

58 "yaml", 

59 "--context", 

60 context_arn, 

61 ], 

62 capture_output=True, 

63 text=True, 

64 timeout=_KUBECTL_TIMEOUT_SECONDS, 

65 ) 

66 except FileNotFoundError: 

67 return json.dumps({"error": "kubectl not found"}) 

68 except cli_runner.subprocess.TimeoutExpired: # type: ignore[attr-defined] 

69 return json.dumps({"error": f"kubectl timed out after {_KUBECTL_TIMEOUT_SECONDS}s"}) 

70 if result.returncode != 0: 

71 err = (result.stderr or result.stdout or "").strip() 

72 return json.dumps( 

73 {"error": err or "kubectl command failed", "exit_code": result.returncode} 

74 ) 

75 return str(result.stdout) 

76 

77 

78def register(mcp_instance: Any) -> None: 

79 """Register regional live job-state resources and the fail-closed legacy URI.""" 

80 mcp_instance.resource("gco://jobs/{job_name}")(_job_resource) 

81 mcp_instance.resource("gco://jobs/{region}/{job_name}")(_job_resource_for_region)