Coverage for cli / gitops.py: 100.00%
50 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""The self-managed Argo CD behind ``gco gitops`` and the ``gitops_status`` MCP tool.
3GCO installs Argo CD into each regional cluster from the upstream chart when
4``cdk.json`` ``helm.argocd.enabled`` is true (see ``gco/argocd_config.py`` and
5``docs/GITOPS.md``). This module holds the command logic:
7* :func:`build_status` — what cdk.json says about Argo CD (enabled, the pinned
8 chart, the fence, the GitOps hand-off rendered per Region, the repo-server
9 size or autoscaler) plus how to reach the UI. Pure: reads cdk.json and
10 charts.yaml only, so it works anywhere.
11* :func:`read_admin_password` — the generated ``admin`` password from
12 ``argocd-initial-admin-secret`` (through the tunnelled kubectl session).
13* :func:`create_session_token` / :func:`capture_argocd_screenshot` — log in to
14 the Argo CD API over the port-forward and screenshot the Applications view
15 with the session cookie, so a capture needs no human.
17Imports nothing from ``aws_cdk``.
18"""
20from __future__ import annotations
22from collections.abc import Mapping
23from pathlib import Path
24from typing import Any
26from gco.argocd_config import (
27 ARGOCD_ADMIN_USERNAME,
28 ARGOCD_CHART_NAME,
29 ARGOCD_HELM_KEY,
30 ARGOCD_INITIAL_ADMIN_SECRET,
31 ARGOCD_NAMESPACE,
32 ARGOCD_REPO_SERVER_DEPLOYMENT,
33 ARGOCD_SERVER_PORT,
34 ARGOCD_SERVER_SERVICE,
35 GITOPS_DEFAULT_NAMESPACE,
36 GITOPS_PROJECT_NAME,
37 GITOPS_ROOT_APPLICATION_NAME,
38 GITOPS_TENANT_NAMESPACES,
39 gitops_enabled,
40 render_gitops_path,
41 validate_argocd_config,
42)
44from . import cluster_ui
46#: Default local port for ``gco gitops open`` / ``screenshot`` (Argo CD's
47#: customary local port; Grafana keeps 3000, MLflow 5000).
48DEFAULT_LOCAL_PORT = 8080
50#: Default screenshot filename; ``docs/GITOPS.md`` embeds ``images/argocd-ui.png``.
51DEFAULT_SCREENSHOT_FILENAME = "argocd-ui.png"
53#: The Argo CD UI route a capture lands on.
54APPLICATIONS_PATH = "/applications"
56#: Cookie the Argo CD UI reads its session token from.
57SESSION_COOKIE = "argocd.token"
60def load_argocd_config(cdk_json_path: Path | None = None) -> dict[str, Any]:
61 """The validated ``helm.argocd`` block of cdk.json (defaults filled in).
63 Raises ``RuntimeError`` when cdk.json is missing and
64 ``gco.argocd_config.ArgoCdConfigError`` (a ``ValueError``) when the block
65 is malformed — the same rules the CDK app applies.
66 """
67 context = cluster_ui.load_cdk_context(cdk_json_path)
68 helm = context.get("helm")
69 return validate_argocd_config(helm.get(ARGOCD_HELM_KEY) if isinstance(helm, dict) else None)
72def build_status(
73 config: Mapping[str, Any],
74 *,
75 regions: list[str],
76 project_name: str,
77 chart: Mapping[str, Any] | None = None,
78) -> dict[str, Any]:
79 """The configured Argo CD for ``regions`` as one document.
81 ``gitops.paths`` renders the per-cluster path placeholders for every
82 Region, so an operator sees exactly which directory each cluster syncs.
83 ``enabled`` is the cdk.json toggle; a run-scoped
84 ``helm_enabled_overrides=argocd`` deploy is not visible here.
85 """
86 document: dict[str, Any] = {
87 "enabled": bool(config["enabled"]),
88 "chart": dict(chart) if chart else None,
89 "namespace": ARGOCD_NAMESPACE,
90 "project": GITOPS_PROJECT_NAME,
91 "destinations": list(GITOPS_TENANT_NAMESPACES),
92 "source_repos": list(config["source_repos"]),
93 }
94 if gitops_enabled(config):
95 gitops = config["gitops"]
96 document["gitops"] = {
97 "enabled": True,
98 "application": GITOPS_ROOT_APPLICATION_NAME,
99 "repo_url": str(gitops["repo_url"]).strip(),
100 "revision": str(gitops["revision"]).strip(),
101 "path": str(gitops["path"]),
102 "paths": {
103 region: render_gitops_path(
104 str(gitops["path"]), region=region, cluster_name=f"{project_name}-{region}"
105 )
106 for region in regions
107 },
108 "default_namespace": GITOPS_DEFAULT_NAMESPACE,
109 "sync_policy": str(gitops["sync_policy"]),
110 }
111 else:
112 document["gitops"] = {"enabled": False}
113 repo_server = config["repo_server"]
114 autoscaling = repo_server["autoscaling"]
115 document["repo_server"] = {
116 "deployment": ARGOCD_REPO_SERVER_DEPLOYMENT,
117 "replicas": int(repo_server["replicas"]),
118 "autoscaling": (
119 {
120 "enabled": True,
121 "min_replicas": int(repo_server["replicas"]),
122 "max_replicas": int(autoscaling["max_replicas"]),
123 "cpu_target_utilization_percentage": int(
124 autoscaling["cpu_target_utilization_percentage"]
125 ),
126 }
127 if autoscaling["enabled"]
128 else {"enabled": False}
129 ),
130 }
131 document["ui"] = {
132 "service": f"svc/{ARGOCD_SERVER_SERVICE}",
133 "open": "gco gitops open",
134 "username": ARGOCD_ADMIN_USERNAME,
135 "password": "gco gitops password", # nosec B105 # the command that prints it, not a credential
136 }
137 return document
140def gitops_status(
141 *,
142 regions: list[str],
143 project_name: str,
144 cdk_json_path: Path | None = None,
145 charts_yaml: Path | None = None,
146) -> dict[str, Any]:
147 """:func:`build_status` over the working directory's cdk.json and charts.yaml."""
148 return build_status(
149 load_argocd_config(cdk_json_path),
150 regions=regions,
151 project_name=project_name,
152 chart=cluster_ui.chart_entry(ARGOCD_CHART_NAME, charts_yaml),
153 )
156def read_admin_password(*, server: str | None = None, tls_server_name: str | None = None) -> str:
157 """The generated ``admin`` password Argo CD wrote on first start.
159 Argo CD recommends deleting ``argocd-initial-admin-secret`` once the
160 password is changed; a missing Secret is reported with that remedy.
161 """
162 try:
163 return cluster_ui.read_secret_key(
164 ARGOCD_NAMESPACE,
165 ARGOCD_INITIAL_ADMIN_SECRET,
166 "password",
167 server=server,
168 tls_server_name=tls_server_name,
169 )
170 except RuntimeError as exc:
171 # kubectl reports a missing Secret (or a missing argocd namespace) as
172 # "Error from server (NotFound): ..."; "kubectl not found" is not that.
173 if "(NotFound)" in str(exc):
174 raise RuntimeError(
175 f"{ARGOCD_NAMESPACE}/{ARGOCD_INITIAL_ADMIN_SECRET} is gone: the initial admin "
176 "password was rotated (or Argo CD is not installed; check 'gco gitops status'). "
177 "Use the current admin password."
178 ) from exc
179 raise
182def create_session_token(base_url: str, password: str, *, timeout: float = 30.0) -> str:
183 """Log in to the Argo CD API as ``admin`` and return the session token."""
184 import requests
186 try:
187 response = requests.post(
188 f"{base_url.rstrip('/')}/api/v1/session",
189 json={"username": ARGOCD_ADMIN_USERNAME, "password": password},
190 timeout=timeout,
191 )
192 except requests.RequestException as exc:
193 raise RuntimeError(f"Could not reach the Argo CD API at {base_url}: {exc}") from exc
194 if response.status_code != 200:
195 raise RuntimeError(
196 f"Argo CD refused the admin login ({response.status_code}); "
197 "pass the current password with --password"
198 )
199 token = (response.json() or {}).get("token")
200 if not isinstance(token, str) or not token:
201 raise RuntimeError("Argo CD returned no session token")
202 return token
205def capture_argocd_screenshot(
206 base_url: str,
207 output: Path,
208 *,
209 token: str,
210 headless: bool = True,
211) -> Path:
212 """Screenshot the Applications view with an authenticated session cookie."""
213 base = base_url.rstrip("/")
214 return cluster_ui.capture_page(
215 f"{base}{APPLICATIONS_PATH}",
216 output,
217 cookies=[{"name": SESSION_COOKIE, "value": token, "url": base}],
218 headless=headless,
219 )
222#: What ``open`` forwards (namespace, target, remote port).
223PORT_FORWARD_TARGET = (ARGOCD_NAMESPACE, f"svc/{ARGOCD_SERVER_SERVICE}", ARGOCD_SERVER_PORT)