Coverage for cli / commands / gitops_cmd.py: 100.00%
103 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""GCO GitOps command group — the self-managed Argo CD.
3Provides the ``gco gitops`` sub-commands:
5* ``status`` — what cdk.json configures (``helm.argocd``): enabled, the pinned
6 chart, the gco-tenants fence, the GitOps hand-off rendered per Region and
7 the repo-server size or autoscaler.
8* ``open`` — ``kubectl port-forward`` to the Argo CD UI over the PRIVATE EKS
9 API endpoint (``--via-ssm`` tunnels through an SSM-managed instance), like
10 ``gco monitoring open``.
11* ``password`` — the generated ``admin`` password from
12 ``argocd-initial-admin-secret``.
13* ``screenshot`` — a headless capture of the Applications view (API login,
14 session cookie, full-page PNG).
16See docs/GITOPS.md. In-cluster access always goes through the private API
17endpoint — there is no public Argo CD ingress.
18"""
20from __future__ import annotations
22import sys
23from pathlib import Path
24from typing import Any
26import click
28from .. import cluster_ui
29from ..config import GCOConfig
30from ..output import get_output_formatter
32pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
35@click.group()
36@pass_config
37def gitops(config: Any) -> None:
38 """Self-managed Argo CD (cdk.json helm.argocd): GitOps into the tenant namespaces."""
41@gitops.command("status")
42@click.option("--region", "-r", help="Render the GitOps path for this region only.")
43@pass_config
44def gitops_status(config: Any, region: str | None) -> None:
45 """Show the configured Argo CD: toggle, chart pin, fence, GitOps hand-off, repo server.
47 Reads cdk.json (helm.argocd) and charts.yaml only — no AWS or cluster
48 access. The GitOps path is rendered for every regional deployment region
49 (or just --region).
51 Examples:
52 gco gitops status
53 gco gitops status -r us-west-2 --output json
54 """
55 from ..gitops import gitops_status as _gitops_status
57 formatter = get_output_formatter(config)
58 try:
59 regions = (
60 [str(region)] if region else cluster_ui.regional_regions(cluster_ui.load_cdk_context())
61 )
62 status = _gitops_status(regions=regions, project_name=config.project_name)
63 except (RuntimeError, ValueError) as exc:
64 formatter.print_error(f"Failed to read helm.argocd from cdk.json: {exc}")
65 sys.exit(1)
66 formatter.print(status)
67 if not status["enabled"]:
68 formatter.print_info(
69 "Argo CD is off (cdk.json helm.argocd.enabled); enable it and run "
70 "'gco stacks deploy' (or deploy once with '--enable argocd')."
71 )
74@gitops.command("open")
75@cluster_ui.tunnel_options
76@click.option("--local-port", type=int, help="Local port to bind (default 8080).", default=None)
77@pass_config
78def gitops_open(
79 config: Any,
80 region: str | None,
81 via_ssm: str | None,
82 bastion_ttl_minutes: int,
83 assume_yes: bool,
84 local_port: int | None,
85) -> None:
86 """Port-forward to the Argo CD UI over the private EKS endpoint.
88 Runs in the foreground; press Ctrl-C to stop. Sign in as admin with the
89 password from 'gco gitops password'.
91 Examples:
92 gco gitops open --via-ssm auto -y
93 gco gitops open -r us-west-2 --local-port 18080 --via-ssm i-0123456789abcdef0
94 """
95 from ..cluster_tunnel import open_api_server_tunnel, resolve_region
96 from ..gitops import DEFAULT_LOCAL_PORT, PORT_FORWARD_TARGET
97 from ..kubectl_helpers import build_port_forward_command, update_kubeconfig
99 formatter = get_output_formatter(config)
100 target_region = resolve_region(config, region)
101 cluster = f"{config.project_name}-{target_region}"
102 bind_port = local_port or DEFAULT_LOCAL_PORT
103 namespace, target, remote_port = PORT_FORWARD_TARGET
104 try:
105 update_kubeconfig(cluster, target_region)
106 with open_api_server_tunnel(
107 formatter,
108 cluster=cluster,
109 region=target_region,
110 via_ssm=via_ssm,
111 bastion_ttl_minutes=bastion_ttl_minutes,
112 assume_yes=assume_yes,
113 ) as session:
114 cmd = build_port_forward_command(
115 namespace,
116 target,
117 bind_port,
118 remote_port,
119 server=session.server,
120 tls_server_name=session.tls_server_name,
121 )
122 formatter.print_info(
123 f"Starting kubectl port-forward for Argo CD on localhost:{bind_port}..."
124 )
126 def _announce() -> None:
127 formatter.print_success(
128 f"Forwarding Argo CD → http://localhost:{bind_port} (Ctrl-C to stop)"
129 )
130 formatter.print_info(
131 "Sign in as admin; 'gco gitops password' prints the generated password."
132 )
134 try:
135 cluster_ui.exec_port_forward(cmd, bind_port, on_ready=_announce)
136 except KeyboardInterrupt: # pragma: no cover - interactive Ctrl-C
137 return
138 except (RuntimeError, ValueError) as exc:
139 formatter.print_error(str(exc))
140 sys.exit(1)
143@gitops.command("password")
144@cluster_ui.tunnel_options
145@pass_config
146def gitops_password(
147 config: Any,
148 region: str | None,
149 via_ssm: str | None,
150 bastion_ttl_minutes: int,
151 assume_yes: bool,
152) -> None:
153 """Print the generated Argo CD admin password (argocd-initial-admin-secret).
155 Argo CD writes it on first start. Once you change the admin password,
156 delete that Secret; this command then says so.
158 Examples:
159 gco gitops password --via-ssm auto -y
160 """
161 from ..cluster_tunnel import open_api_server_tunnel, resolve_region
162 from ..gitops import read_admin_password
163 from ..kubectl_helpers import update_kubeconfig
165 formatter = get_output_formatter(config)
166 target_region = resolve_region(config, region)
167 cluster = f"{config.project_name}-{target_region}"
168 try:
169 update_kubeconfig(cluster, target_region)
170 with open_api_server_tunnel(
171 formatter,
172 cluster=cluster,
173 region=target_region,
174 via_ssm=via_ssm,
175 bastion_ttl_minutes=bastion_ttl_minutes,
176 assume_yes=assume_yes,
177 ) as session:
178 password = read_admin_password(
179 server=session.server, tls_server_name=session.tls_server_name
180 )
181 except (RuntimeError, ValueError) as exc:
182 formatter.print_error(str(exc))
183 sys.exit(1)
184 formatter.print({"region": target_region, "username": "admin", "password": password})
187@gitops.command("screenshot")
188@cluster_ui.tunnel_options
189@click.option(
190 "--output",
191 "-o",
192 "output_path",
193 type=click.Path(dir_okay=False, path_type=Path),
194 default=None,
195 help="PNG to write (default: ./argocd-ui.png).",
196)
197@click.option("--local-port", type=int, help="Local port to bind (default 8080).", default=None)
198@click.option(
199 "--password",
200 envvar="GCO_ARGOCD_ADMIN_PASSWORD",
201 help=(
202 "Admin password (also $GCO_ARGOCD_ADMIN_PASSWORD; default: read from "
203 "argocd-initial-admin-secret)."
204 ),
205)
206@click.option("--headed", is_flag=True, help="Show the browser window while capturing.")
207@pass_config
208def gitops_screenshot(
209 config: Any,
210 region: str | None,
211 via_ssm: str | None,
212 bastion_ttl_minutes: int,
213 assume_yes: bool,
214 output_path: Path | None,
215 local_port: int | None,
216 password: str | None,
217 headed: bool,
218) -> None:
219 """Capture the Argo CD Applications view as a full-page PNG.
221 Forwards the UI in the background, logs in to the Argo CD API as admin and
222 screenshots /applications with the session cookie — no browser sign-in.
223 Needs Playwright: pip install 'gco[diagrams]' && playwright install chromium.
225 Examples:
226 gco gitops screenshot --via-ssm auto -y
227 gco gitops screenshot --output images/argocd-ui.png
228 """
229 from ..cluster_tunnel import open_api_server_tunnel, resolve_region
230 from ..cluster_ui import background_port_forward
231 from ..gitops import (
232 DEFAULT_LOCAL_PORT,
233 DEFAULT_SCREENSHOT_FILENAME,
234 PORT_FORWARD_TARGET,
235 capture_argocd_screenshot,
236 create_session_token,
237 read_admin_password,
238 )
239 from ..kubectl_helpers import build_port_forward_command, update_kubeconfig
241 formatter = get_output_formatter(config)
242 target_region = resolve_region(config, region)
243 cluster = f"{config.project_name}-{target_region}"
244 bind_port = local_port or DEFAULT_LOCAL_PORT
245 output = output_path or Path.cwd() / DEFAULT_SCREENSHOT_FILENAME
246 namespace, target, remote_port = PORT_FORWARD_TARGET
247 base_url = f"http://localhost:{bind_port}"
248 try:
249 update_kubeconfig(cluster, target_region)
250 with open_api_server_tunnel(
251 formatter,
252 cluster=cluster,
253 region=target_region,
254 via_ssm=via_ssm,
255 bastion_ttl_minutes=bastion_ttl_minutes,
256 assume_yes=assume_yes,
257 ) as session:
258 admin_password = password or read_admin_password(
259 server=session.server, tls_server_name=session.tls_server_name
260 )
261 cmd = build_port_forward_command(
262 namespace,
263 target,
264 bind_port,
265 remote_port,
266 server=session.server,
267 tls_server_name=session.tls_server_name,
268 )
269 with background_port_forward(cmd, bind_port):
270 token = create_session_token(base_url, admin_password)
271 written = capture_argocd_screenshot(
272 base_url, output, token=token, headless=not headed
273 )
274 except (RuntimeError, ValueError) as exc:
275 formatter.print_error(str(exc))
276 sys.exit(1)
277 formatter.print_success(f"Saved the Argo CD Applications view to {written}")
278 formatter.print({"region": target_region, "output": str(written)})