Coverage for cli / crossplane.py: 100.00%

1"""The self-managed Crossplane behind ``gco crossplane`` and the ``crossplane_status`` MCP tool. 

2 

3GCO installs Crossplane and its Crossview dashboard into each regional cluster 

4from the upstream charts when ``cdk.json`` ``helm.crossplane.enabled`` is true 

5(see ``docs/CROSSPLANE.md``). This module holds the command logic: 

6 

7* :func:`build_status` — what cdk.json says (enabled), the pinned charts, the 

8 composition functions GCO installs (read from the shipped 

9 ``post-helm-crossplane.yaml``) and how to reach the dashboard. Pure: reads 

10 cdk.json, charts.yaml and the manifest only, so it works anywhere. 

11* :func:`capture_dashboard_screenshot` — a headless capture of Crossview over 

12 the port-forward (the dashboard runs with auth mode none behind it). 

13 

14Imports nothing from ``aws_cdk``. 

15""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Mapping 

20from pathlib import Path 

21from typing import Any 

22 

23from . import cluster_ui 

24 

25#: ``cdk.json`` ``helm`` block key (and ``helm_enabled_overrides`` name). 

26CROSSPLANE_HELM_KEY = "crossplane" 

27 

28#: charts.yaml entries the one toggle installs, in install order. 

29CROSSPLANE_CHARTS: tuple[str, ...] = ("crossplane", "crossview") 

30 

31CROSSPLANE_NAMESPACE = "crossplane-system" 

32 

33#: The GCO-owned post-Helm manifest carrying the composition functions. 

34CROSSPLANE_MANIFEST = "post-helm-crossplane.yaml" 

35 

36#: The dashboard Service (chart release ``crossview``) and its port. 

37DASHBOARD_SERVICE = "crossview-service" 

38DASHBOARD_PORT = 80 

39 

40#: Default local port for ``gco crossplane open`` / ``screenshot`` (Crossview's 

41#: own container port; Grafana keeps 3000, Argo CD 8080). 

42DEFAULT_LOCAL_PORT = 3001 

43 

44#: Default screenshot filename; ``docs/CROSSPLANE.md`` embeds 

45#: ``images/crossview-dashboard.png``. 

46DEFAULT_SCREENSHOT_FILENAME = "crossview-dashboard.png" 

47 

48#: What ``open`` forwards (namespace, target, remote port). 

49PORT_FORWARD_TARGET = (CROSSPLANE_NAMESPACE, f"svc/{DASHBOARD_SERVICE}", DASHBOARD_PORT) 

50 

51 

52def crossplane_enabled(context: Mapping[str, Any]) -> bool: 

53 """The ``helm.crossplane.enabled`` toggle (absent means off, like the stack).""" 

54 helm = context.get("helm") 

55 block = helm.get(CROSSPLANE_HELM_KEY) if isinstance(helm, Mapping) else None 

56 if block is None: 

57 return False 

58 if not isinstance(block, Mapping) or type(block.get("enabled", False)) is not bool: 

59 raise ValueError(f"helm.crossplane must be an object with a boolean enabled, got {block!r}") 

60 return bool(block.get("enabled", False)) 

61 

62 

63def shipped_functions(manifests_dir: Path | None = None) -> list[dict[str, str]]: 

64 """The ``Function`` packages ``post-helm-crossplane.yaml`` installs.""" 

65 return [ 

66 { 

67 "name": str((doc.get("metadata") or {}).get("name") or ""), 

68 "package": str((doc.get("spec") or {}).get("package") or ""), 

69 } 

70 for doc in cluster_ui.manifest_documents(CROSSPLANE_MANIFEST, manifests_dir) 

71 if doc.get("kind") == "Function" 

72 ] 

73 

74 

75def build_status( 

76 *, 

77 enabled: bool, 

78 charts: list[dict[str, Any]], 

79 functions: list[dict[str, str]], 

80) -> dict[str, Any]: 

81 """The configured Crossplane as one document.""" 

82 return { 

83 "enabled": enabled, 

84 "charts": charts, 

85 "namespace": CROSSPLANE_NAMESPACE, 

86 "functions": functions, 

87 "dashboard": { 

88 "name": "Crossview", 

89 "service": f"svc/{DASHBOARD_SERVICE}", 

90 "open": "gco crossplane open", 

91 "auth": "none (reachable only through the kubectl port-forward)", 

92 }, 

93 } 

94 

95 

96def crossplane_status( 

97 *, 

98 cdk_json_path: Path | None = None, 

99 charts_yaml: Path | None = None, 

100 manifests_dir: Path | None = None, 

101) -> dict[str, Any]: 

102 """:func:`build_status` over the working directory's cdk.json and the shipped files.""" 

103 context = cluster_ui.load_cdk_context(cdk_json_path) 

104 charts = [ 

105 entry 

106 for entry in (cluster_ui.chart_entry(name, charts_yaml) for name in CROSSPLANE_CHARTS) 

107 if entry is not None 

108 ] 

109 return build_status( 

110 enabled=crossplane_enabled(context), 

111 charts=charts, 

112 functions=shipped_functions(manifests_dir), 

113 ) 

114 

115 

116def capture_dashboard_screenshot(base_url: str, output: Path, *, headless: bool = True) -> Path: 

117 """Screenshot the Crossview dashboard's landing page.""" 

118 return cluster_ui.capture_page(f"{base_url.rstrip('/')}/", output, headless=headless)