Coverage for scripts / upgrade_validation / __main__.py: 100.00%
122 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""CLI entry: ``python -m scripts.upgrade_validation``.
3Mirrors ``scripts.live_release_validation.__main__`` (the same identity flags,
4consent posture, and checkpoint/resume semantics) plus the release to upgrade
5from (``--base-ref``, by default the newest release tag in the checkout's
6history) and the private workspace the base release is cloned into.
7"""
9from __future__ import annotations
11import argparse
12import contextlib
13import os
14import re
15import subprocess
16import sys
17import traceback
18from datetime import UTC, datetime
19from pathlib import Path
21from cli.upgrade import RELEASE_TAG_RE, parse_release_tags
22from scripts.live_release_validation.cli_args import (
23 path_from_root,
24 repository_root,
25 split_csv_names,
26)
27from scripts.live_release_validation.models import ValidationReport, utc_now
28from scripts.live_release_validation.runner import (
29 LiveValidationRunner,
30 require_local_execution,
31)
33from .models import UpgradeRunSettings
34from .registry import build_action_registry
35from .workspace import Workspace, remove_workspace
37REPORT_TITLE = "GCO Upgrade Validation"
38REPORT_STEM = "upgrade-validation"
39DEFAULT_REPORTS_DIR = Path("~/gco-upgrade-validation-reports")
42def _build_parser() -> argparse.ArgumentParser:
43 registry = build_action_registry()
44 parser = argparse.ArgumentParser(
45 prog="python -m scripts.upgrade_validation",
46 description=(
47 "Deploy the previous release with its own gco, upgrade it to this checkout "
48 "with that gco's upgrade command, verify the upgraded deployment, and always "
49 "destroy what was deployed. Reports carry account-specific identifiers; post "
50 "only sanitized summaries."
51 ),
52 )
53 parser.add_argument("--repo-root", help="GCO checkout (default: current Git root)")
54 parser.add_argument(
55 "--expected-account",
56 default=os.environ.get("GCO_LIVE_EXPECTED_ACCOUNT"),
57 help="Exact 12-digit AWS account ID (or GCO_LIVE_EXPECTED_ACCOUNT)",
58 )
59 parser.add_argument(
60 "--expected-sha",
61 default=os.environ.get("GCO_LIVE_EXPECTED_SHA"),
62 help="Exact 40-character Git commit (or GCO_LIVE_EXPECTED_SHA)",
63 )
64 parser.add_argument(
65 "--expected-branch",
66 default=os.environ.get("GCO_LIVE_EXPECTED_BRANCH"),
67 help="Exact local branch identity (or GCO_LIVE_EXPECTED_BRANCH)",
68 )
69 parser.add_argument(
70 "--base-ref",
71 metavar="vX.Y.Z",
72 help="Release to deploy first and upgrade from (default: the newest earlier release)",
73 )
74 parser.add_argument(
75 "--actions",
76 type=split_csv_names,
77 default=("all",),
78 metavar="NAME[,NAME...]",
79 help="Selectable actions; dependencies are added automatically (default: all)",
80 )
81 parser.add_argument("--list-actions", action="store_true", help="List actions and exit")
82 parser.add_argument("--run-id", help="Stable run/checkpoint identifier")
83 parser.add_argument(
84 "--report-dir", help=f"Report directory (default: {DEFAULT_REPORTS_DIR}/<run-id>)"
85 )
86 parser.add_argument(
87 "--workspace-dir",
88 help="Private workspace for the base release (default: <report-dir>.workspace)",
89 )
90 parser.add_argument(
91 "--checkpoint", help="Checkpoint JSON path (default: <report-dir>/checkpoint.json)"
92 )
93 parser.add_argument(
94 "--resume", action="store_true", help="Resume an exact identity-matched checkpoint"
95 )
96 parser.add_argument(
97 "--protected-stack",
98 action="append",
99 default=[],
100 metavar="NAME",
101 help="Additional non-project CloudFormation stack to preserve exactly",
102 )
103 parser.add_argument(
104 "--confirm-kms-key-deletion",
105 action="store_true",
106 help=(
107 "Explicitly authorize scheduling only this run's exact retained EKS KMS keys "
108 "(both generations) for deletion after stack teardown"
109 ),
110 )
111 parser.add_argument(
112 "--min-free-disk-gib",
113 type=int,
114 default=20,
115 metavar="GIB",
116 help="Free-space floor for the checkout, reports, home, and workspace (0 disables)",
117 )
118 parser.epilog = "Actions: " + ", ".join(registry)
119 return parser
122def _git_output(root: Path, *arguments: str) -> str:
123 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed git argv, no shell
124 ["git", *arguments], cwd=root, capture_output=True, text=True, check=False
125 )
126 return result.stdout.strip() if result.returncode == 0 else ""
129def _tag_commit(root: Path, tag: str) -> str:
130 return _git_output(root, "rev-parse", "--verify", "--quiet", f"refs/tags/{tag}^{{commit}}")
133def _resolve_base(
134 parser: argparse.ArgumentParser, root: Path, requested: str | None, expected_sha: str
135) -> tuple[str, str]:
136 """The base release tag and its commit: the named one, or the newest earlier release."""
137 if requested:
138 if not RELEASE_TAG_RE.fullmatch(requested):
139 parser.error("--base-ref must name a release tag vMAJOR.MINOR.PATCH")
140 commit = _tag_commit(root, requested)
141 if not commit:
142 parser.error(f"--base-ref {requested} is not a tag in {root}")
143 return requested, commit
144 merged = _git_output(root, "tag", "--list", "v*", "--merged", expected_sha).splitlines()
145 for tag in reversed(parse_release_tags(merged)):
146 commit = _tag_commit(root, tag.name)
147 if commit and commit != expected_sha:
148 return tag.name, commit
149 parser.error(f"No release tag precedes {expected_sha} in {root}; pass --base-ref")
152def _settings_from_args(
153 parser: argparse.ArgumentParser, args: argparse.Namespace
154) -> UpgradeRunSettings:
155 if not args.expected_account or not re.fullmatch(r"\d{12}", args.expected_account):
156 parser.error("--expected-account must be an exact 12-digit AWS account ID")
157 if not args.expected_sha or not re.fullmatch(r"[0-9a-fA-F]{40}", args.expected_sha):
158 parser.error("--expected-sha must be an exact 40-character commit SHA")
159 if not args.expected_branch or not args.expected_branch.strip():
160 parser.error("--expected-branch is required")
161 if args.run_id and not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}", args.run_id):
162 parser.error("--run-id must be 1-80 safe filename characters")
163 if args.min_free_disk_gib < 0:
164 parser.error("--min-free-disk-gib must be >= 0 (0 disables the check)")
165 root = repository_root(args.repo_root)
166 expected_sha = args.expected_sha.lower()
167 base_ref, base_commit = _resolve_base(parser, root, args.base_ref, expected_sha)
168 run_id = args.run_id or (datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + "-" + expected_sha[:12])
169 report_dir = path_from_root(root, args.report_dir, DEFAULT_REPORTS_DIR.expanduser() / run_id)
170 workspace_dir = path_from_root(
171 root, args.workspace_dir, report_dir.with_name(f"{report_dir.name}.workspace")
172 )
173 checkpoint = path_from_root(root, args.checkpoint, report_dir / "checkpoint.json")
174 protected = tuple(dict.fromkeys(("CDKToolkit", "GCOGitHubOIDCStack", *args.protected_stack)))
175 return UpgradeRunSettings(
176 run_id=run_id,
177 repo_root=root,
178 report_dir=report_dir,
179 checkpoint_path=checkpoint,
180 expected_account=args.expected_account,
181 expected_sha=expected_sha,
182 expected_branch=args.expected_branch.strip(),
183 profile="configured",
184 requested_actions=args.actions,
185 protected_stack_names=protected,
186 confirm_kms_key_deletion=args.confirm_kms_key_deletion,
187 resume=args.resume,
188 min_free_disk_gib=args.min_free_disk_gib,
189 base_ref=base_ref,
190 base_commit=base_commit,
191 workspace_dir=workspace_dir,
192 )
195def _remove_finished_workspace(settings: UpgradeRunSettings, runner: LiveValidationRunner) -> None:
196 """Delete the workspace once teardown completed; it has no use after that."""
197 if not runner.checkpoint.destroyed:
198 print(f"Workspace kept for resume or inspection: {settings.workspace_dir}")
199 return
200 try:
201 removed = remove_workspace(Workspace(settings.workspace_dir), settings.run_id)
202 except (OSError, RuntimeError) as exc:
203 print(f"Could not remove the workspace {settings.workspace_dir}: {exc}", file=sys.stderr)
204 return
205 if removed:
206 print(f"Removed the workspace {settings.workspace_dir}")
209def main() -> int:
210 parser = _build_parser()
211 args = parser.parse_args()
212 if args.list_actions:
213 for definition in build_action_registry().values():
214 dependencies = ", ".join(definition.dependencies) or "none"
215 print(f"{definition.name:16} {definition.description} [depends: {dependencies}]")
216 return 0
217 try:
218 require_local_execution()
219 except RuntimeError as exc:
220 print(f"Upgrade validation could not start: {exc}", file=sys.stderr)
221 return 1
222 settings: UpgradeRunSettings | None = None
223 try:
224 settings = _settings_from_args(parser, args)
225 runner = LiveValidationRunner(settings, registry=build_action_registry())
226 runner.report.title = REPORT_TITLE
227 runner.report.report_stem = REPORT_STEM
228 code = runner.run()
229 _remove_finished_workspace(settings, runner)
230 return code
231 except KeyboardInterrupt:
232 print("Upgrade validation interrupted before the runner initialized", file=sys.stderr)
233 return 130
234 except BaseException as exc:
235 print(f"Upgrade validation could not start: {type(exc).__name__}: {exc}", file=sys.stderr)
236 if settings is not None:
237 report = ValidationReport(
238 run_id=settings.run_id,
239 identity=settings.identity(),
240 selected_actions=list(settings.requested_actions),
241 started_at=utc_now(),
242 ended_at=utc_now(),
243 status="failed",
244 fatal_error="".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
245 title=REPORT_TITLE,
246 report_stem=REPORT_STEM,
247 )
248 with contextlib.suppress(OSError):
249 report.write(settings.report_dir)
250 return 1
253if __name__ == "__main__":
254 sys.exit(main())