Coverage for cli / commands / crossplane_cmd.py: 100.00%

1"""GCO Crossplane command group — the self-managed Crossplane and its Crossview dashboard. 

2 

3Provides the ``gco crossplane`` sub-commands: 

4 

5* ``status`` — what cdk.json configures (``helm.crossplane``): enabled, the 

6 pinned Crossplane and Crossview charts, and the composition functions GCO 

7 installs. 

8* ``open`` — ``kubectl port-forward`` to the Crossview dashboard over the 

9 PRIVATE EKS API endpoint (``--via-ssm`` tunnels through an SSM-managed 

10 instance), like ``gco monitoring open``. 

11* ``screenshot`` — a headless capture of the dashboard (full-page PNG). 

12 

13See docs/CROSSPLANE.md. In-cluster access always goes through the private API 

14endpoint — there is no public dashboard ingress. 

15""" 

16 

17from __future__ import annotations 

18 

19import sys 

20from pathlib import Path 

21from typing import Any 

22 

23import click 

24 

25from .. import cluster_ui 

26from ..config import GCOConfig 

27from ..output import get_output_formatter 

28 

29pass_config = click.make_pass_decorator(GCOConfig, ensure=True) 

30 

31 

32@click.group() 

33@pass_config 

34def crossplane(config: Any) -> None: 

35 """Self-managed Crossplane (cdk.json helm.crossplane) and its Crossview dashboard.""" 

36 

37 

38@crossplane.command("status") 

39@pass_config 

40def crossplane_status(config: Any) -> None: 

41 """Show the configured Crossplane: toggle, chart pins and composition functions. 

42 

43 Reads cdk.json (helm.crossplane), charts.yaml and the shipped post-Helm 

44 manifest only — no AWS or cluster access. 

45 

46 Examples: 

47 gco crossplane status 

48 gco crossplane status --output json 

49 """ 

50 from ..crossplane import crossplane_status as _crossplane_status 

51 

52 formatter = get_output_formatter(config) 

53 try: 

54 status = _crossplane_status() 

55 except (RuntimeError, ValueError) as exc: 

56 formatter.print_error(f"Failed to read helm.crossplane from cdk.json: {exc}") 

57 sys.exit(1) 

58 formatter.print(status) 

59 if not status["enabled"]: 

60 formatter.print_info( 

61 "Crossplane is off (cdk.json helm.crossplane.enabled); enable it and run " 

62 "'gco stacks deploy' (or deploy once with '--enable crossplane')." 

63 ) 

64 

65 

66@crossplane.command("open") 

67@cluster_ui.tunnel_options 

68@click.option("--local-port", type=int, help="Local port to bind (default 3001).", default=None) 

69@pass_config 

70def crossplane_open( 

71 config: Any, 

72 region: str | None, 

73 via_ssm: str | None, 

74 bastion_ttl_minutes: int, 

75 assume_yes: bool, 

76 local_port: int | None, 

77) -> None: 

78 """Port-forward to the Crossview dashboard over the private EKS endpoint. 

79 

80 Runs in the foreground; press Ctrl-C to stop. The dashboard runs without a 

81 login: the port-forward is the access control. 

82 

83 Examples: 

84 gco crossplane open --via-ssm auto -y 

85 gco crossplane open -r us-west-2 --local-port 13001 --via-ssm i-0123456789abcdef0 

86 """ 

87 from ..cluster_tunnel import open_api_server_tunnel, resolve_region 

88 from ..crossplane import DEFAULT_LOCAL_PORT, PORT_FORWARD_TARGET 

89 from ..kubectl_helpers import build_port_forward_command, update_kubeconfig 

90 

91 formatter = get_output_formatter(config) 

92 target_region = resolve_region(config, region) 

93 cluster = f"{config.project_name}-{target_region}" 

94 bind_port = local_port or DEFAULT_LOCAL_PORT 

95 namespace, target, remote_port = PORT_FORWARD_TARGET 

96 try: 

97 update_kubeconfig(cluster, target_region) 

98 with open_api_server_tunnel( 

99 formatter, 

100 cluster=cluster, 

101 region=target_region, 

102 via_ssm=via_ssm, 

103 bastion_ttl_minutes=bastion_ttl_minutes, 

104 assume_yes=assume_yes, 

105 ) as session: 

106 cmd = build_port_forward_command( 

107 namespace, 

108 target, 

109 bind_port, 

110 remote_port, 

111 server=session.server, 

112 tls_server_name=session.tls_server_name, 

113 ) 

114 formatter.print_info( 

115 f"Starting kubectl port-forward for Crossview on localhost:{bind_port}..." 

116 ) 

117 

118 def _announce() -> None: 

119 formatter.print_success( 

120 f"Forwarding Crossview → http://localhost:{bind_port} (Ctrl-C to stop)" 

121 ) 

122 

123 try: 

124 cluster_ui.exec_port_forward(cmd, bind_port, on_ready=_announce) 

125 except KeyboardInterrupt: # pragma: no cover - interactive Ctrl-C 

126 return 

127 except (RuntimeError, ValueError) as exc: 

128 formatter.print_error(str(exc)) 

129 sys.exit(1) 

130 

131 

132@crossplane.command("screenshot") 

133@cluster_ui.tunnel_options 

134@click.option( 

135 "--output", 

136 "-o", 

137 "output_path", 

138 type=click.Path(dir_okay=False, path_type=Path), 

139 default=None, 

140 help="PNG to write (default: ./crossview-dashboard.png).", 

141) 

142@click.option("--local-port", type=int, help="Local port to bind (default 3001).", default=None) 

143@click.option("--headed", is_flag=True, help="Show the browser window while capturing.") 

144@pass_config 

145def crossplane_screenshot( 

146 config: Any, 

147 region: str | None, 

148 via_ssm: str | None, 

149 bastion_ttl_minutes: int, 

150 assume_yes: bool, 

151 output_path: Path | None, 

152 local_port: int | None, 

153 headed: bool, 

154) -> None: 

155 """Capture the Crossview dashboard as a full-page PNG. 

156 

157 Forwards the dashboard in the background and screenshots its landing 

158 page. Needs Playwright: pip install 'gco[diagrams]' && playwright install 

159 chromium. 

160 

161 Examples: 

162 gco crossplane screenshot --via-ssm auto -y 

163 gco crossplane screenshot --output images/crossview-dashboard.png 

164 """ 

165 from ..cluster_tunnel import open_api_server_tunnel, resolve_region 

166 from ..crossplane import ( 

167 DEFAULT_LOCAL_PORT, 

168 DEFAULT_SCREENSHOT_FILENAME, 

169 PORT_FORWARD_TARGET, 

170 capture_dashboard_screenshot, 

171 ) 

172 from ..kubectl_helpers import build_port_forward_command, update_kubeconfig 

173 

174 formatter = get_output_formatter(config) 

175 target_region = resolve_region(config, region) 

176 cluster = f"{config.project_name}-{target_region}" 

177 bind_port = local_port or DEFAULT_LOCAL_PORT 

178 output = output_path or Path.cwd() / DEFAULT_SCREENSHOT_FILENAME 

179 namespace, target, remote_port = PORT_FORWARD_TARGET 

180 try: 

181 update_kubeconfig(cluster, target_region) 

182 with open_api_server_tunnel( 

183 formatter, 

184 cluster=cluster, 

185 region=target_region, 

186 via_ssm=via_ssm, 

187 bastion_ttl_minutes=bastion_ttl_minutes, 

188 assume_yes=assume_yes, 

189 ) as session: 

190 cmd = build_port_forward_command( 

191 namespace, 

192 target, 

193 bind_port, 

194 remote_port, 

195 server=session.server, 

196 tls_server_name=session.tls_server_name, 

197 ) 

198 with cluster_ui.background_port_forward(cmd, bind_port): 

199 written = capture_dashboard_screenshot( 

200 f"http://localhost:{bind_port}", output, headless=not headed 

201 ) 

202 except (RuntimeError, ValueError) as exc: 

203 formatter.print_error(str(exc)) 

204 sys.exit(1) 

205 formatter.print_success(f"Saved the Crossview dashboard to {written}") 

206 formatter.print({"region": target_region, "output": str(written)})