Coverage for scripts / upgrade_validation / workspace.py: 100.00%
251 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""The run's private workspace: a git mirror, the base-release clone, and its venv.
3Nothing here touches AWS. The workspace holds
5* ``mirror.git``: a bare repository with exactly two refs, the base release
6 tag and the candidate commit (``refs/heads/candidate``). It borrows objects
7 from the operator's repository through git alternates instead of copying
8 them, and it is where the run's synthetic release tag lives, so no tag is
9 ever created in the operator's repository;
10* ``base``: a clone of the mirror checked out at the base tag. Its ``origin``
11 is the mirror, which is how ``gco upgrade`` finds the synthetic tag;
12* ``venv``: the base release's ``gco``, installed editable from the clone
13 with the ``cdk`` extra pinned by the release's own lock file;
14* ``logs``: the full output of every command the harness runs there.
16Every command runs under ``umask 022`` (a restrictive umask makes the CDK
17Lambda assets unreadable, which the deploy refuses) and with the base venv
18first on ``PATH``, because the CDK app runs whichever ``python3`` it finds.
19"""
21from __future__ import annotations
23import hashlib
24import json
25import os
26import re
27import shutil
28import signal
29import subprocess
30import sys
31import threading
32import time
33from collections import deque
34from collections.abc import Callable, Mapping, Sequence
35from dataclasses import dataclass
36from pathlib import Path
37from typing import IO, Any
39from cli.upgrade import ReleaseTag
41#: File whose content (the run id) marks a directory as this run's workspace.
42MARKER_NAME = ".gco-upgrade-validation-run"
43#: Where the private mirror keeps the candidate commit.
44CANDIDATE_REF = "refs/heads/candidate"
45#: Lines of each command's output kept in the report.
46LOG_TAIL_LINES = 40
47#: Seconds a timed-out or interrupted command gets to exit after SIGTERM.
48TERMINATE_GRACE_SECONDS = 60.0
49#: Every workspace command creates files under this umask.
50COMMAND_UMASK = 0o022
52#: Environment the base ``gco`` must not inherit: another interpreter's paths,
53#: an activated venv, git plumbing overrides, and output-format defaults.
54_SCRUBBED_ENVIRONMENT = frozenset(
55 {
56 "PYTHONPATH",
57 "PYTHONHOME",
58 "PYTHONSTARTUP",
59 "PYTHONUSERBASE",
60 "VIRTUAL_ENV",
61 "CONDA_PREFIX",
62 "GIT_DIR",
63 "GIT_WORK_TREE",
64 "GIT_INDEX_FILE",
65 "GIT_OBJECT_DIRECTORY",
66 "GIT_ALTERNATE_OBJECT_DIRECTORIES",
67 "GIT_COMMON_DIR",
68 "GIT_NAMESPACE",
69 "GCO_OUTPUT_FORMAT",
70 }
71)
73Echo = Callable[[str], None]
76@dataclass(frozen=True)
77class Workspace:
78 """Paths inside one run's workspace directory."""
80 root: Path
82 @property
83 def mirror(self) -> Path:
84 return self.root / "mirror.git"
86 @property
87 def clone(self) -> Path:
88 return self.root / "base"
90 @property
91 def venv(self) -> Path:
92 return self.root / "venv"
94 @property
95 def logs(self) -> Path:
96 return self.root / "logs"
98 @property
99 def cache(self) -> Path:
100 return self.root / "gco-cache"
102 @property
103 def marker(self) -> Path:
104 return self.root / MARKER_NAME
106 @property
107 def gco(self) -> Path:
108 return self.venv / "bin" / "gco"
110 @property
111 def python(self) -> Path:
112 return self.venv / "bin" / "python"
114 def owned_by(self, run_id: str) -> bool:
115 """Whether the directory carries this run's marker."""
116 try:
117 return self.marker.read_text(encoding="utf-8").strip() == run_id
118 except OSError:
119 return False
122def reset_workspace(workspace: Workspace, run_id: str) -> None:
123 """Create an empty workspace for this run, clearing its own earlier attempt."""
124 if workspace.root.exists():
125 if not workspace.owned_by(run_id):
126 raise RuntimeError(
127 f"{workspace.root} exists and is not this run's workspace; remove it or "
128 "pass another --workspace-dir"
129 )
130 shutil.rmtree(workspace.root)
131 workspace.root.mkdir(parents=True, mode=0o700)
132 workspace.marker.write_text(f"{run_id}\n", encoding="utf-8")
133 workspace.logs.mkdir(mode=0o700)
136def remove_workspace(workspace: Workspace, run_id: str) -> bool:
137 """Delete this run's workspace; ``False`` when there is none."""
138 if not workspace.root.exists():
139 return False
140 if not workspace.owned_by(run_id):
141 raise RuntimeError(f"Refusing to delete {workspace.root}: it is not this run's workspace")
142 shutil.rmtree(workspace.root)
143 return True
146def _git(cwd: Path, *arguments: str) -> str:
147 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed git argv, no shell
148 ["git", *arguments],
149 cwd=cwd,
150 env=command_environment(None),
151 capture_output=True,
152 text=True,
153 check=False,
154 umask=COMMAND_UMASK,
155 )
156 if result.returncode != 0:
157 message = result.stderr.strip() or result.stdout.strip() or "unknown git error"
158 raise RuntimeError(f"git {' '.join(arguments)} failed in {cwd}: {message}")
159 # Only trailing whitespace: porcelain status lines start with a space.
160 return result.stdout.rstrip()
163def build_mirror(
164 workspace: Workspace,
165 *,
166 source: Path,
167 branch: str,
168 candidate_sha: str,
169 base_ref: str,
170 base_commit: str,
171) -> dict[str, str]:
172 """Create the bare mirror holding only the base tag and the candidate commit."""
173 common_dir = Path(_git(source, "rev-parse", "--path-format=absolute", "--git-common-dir"))
174 _git(workspace.root, "init", "--quiet", "--bare", str(workspace.mirror))
175 alternates = workspace.mirror / "objects" / "info" / "alternates"
176 alternates.write_text(f"{common_dir / 'objects'}\n", encoding="utf-8")
177 _git(
178 workspace.mirror,
179 "fetch",
180 "--quiet",
181 "--no-tags",
182 str(source),
183 f"+refs/tags/{base_ref}:refs/tags/{base_ref}",
184 f"+refs/heads/{branch}:{CANDIDATE_REF}",
185 )
186 fetched_candidate = _git(workspace.mirror, "rev-parse", CANDIDATE_REF)
187 fetched_base = _git(workspace.mirror, "rev-parse", f"refs/tags/{base_ref}^{{commit}}")
188 if fetched_candidate != candidate_sha:
189 raise RuntimeError(
190 f"Branch {branch} is at {fetched_candidate}, not the validated {candidate_sha}"
191 )
192 if fetched_base != base_commit:
193 raise RuntimeError(f"{base_ref} now names {fetched_base}, not {base_commit}")
194 return {"path": str(workspace.mirror), "objects_from": str(common_dir / "objects")}
197def clone_base(workspace: Workspace, *, base_ref: str, base_commit: str) -> dict[str, str]:
198 """Clone the mirror at the base tag (a detached checkout, as a release install is)."""
199 _git(
200 workspace.root,
201 "clone",
202 "--quiet",
203 "--shared",
204 "--branch",
205 base_ref,
206 str(workspace.mirror),
207 str(workspace.clone),
208 )
209 head = checkout_head(workspace)
210 if head != base_commit:
211 raise RuntimeError(f"The base clone is at {head}, not {base_ref} ({base_commit})")
212 return {"path": str(workspace.clone), "head": head}
215def checkout_head(workspace: Workspace) -> str:
216 return _git(workspace.clone, "rev-parse", "HEAD")
219def tracked_changes(workspace: Workspace) -> list[str]:
220 """Tracked paths the clone's working tree changes (untracked files ignored)."""
221 status = _git(workspace.clone, "status", "--porcelain=v1", "--untracked-files=no")
222 return sorted(line[3:] for line in status.splitlines() if line.strip())
225def tag_candidate(workspace: Workspace, *, tag: str, candidate_sha: str) -> dict[str, str]:
226 """Create the synthetic release tag at the candidate commit, in the mirror only."""
227 existing = _git(workspace.mirror, "tag", "--list", tag)
228 if existing:
229 target = _git(workspace.mirror, "rev-parse", f"refs/tags/{tag}^{{commit}}")
230 if target != candidate_sha:
231 raise RuntimeError(f"The mirror's {tag} names {target}, not {candidate_sha}")
232 else:
233 _git(workspace.mirror, "tag", tag, candidate_sha)
234 return {"tag": tag, "commit": candidate_sha, "repository": str(workspace.mirror)}
237def synthetic_release_tag(base_ref: str) -> str:
238 """The next patch release after ``base_ref``: the name the candidate is tagged with.
240 ``gco upgrade --ref`` accepts only release-shaped tags that exist on the
241 remote. The tag is created in the run's private mirror, which holds no
242 other release tag, so it can never collide with a published one.
243 """
244 tag = ReleaseTag.parse(base_ref)
245 if tag is None:
246 raise ValueError(f"{base_ref!r} is not a release tag")
247 major, minor, patch = tag.version
248 return f"v{major}.{minor}.{patch + 1}"
251def sha256_file(path: Path) -> str:
252 return hashlib.sha256(path.read_bytes()).hexdigest()
255def write_run_context(
256 clone: Path,
257 *,
258 run_tag_key: str,
259 run_id: str,
260 context: Mapping[str, str],
261) -> dict[str, Any]:
262 """Add the run tag and the run-scoped context to the clone's cdk.json.
264 The app applies ``context.tags`` to every stack (``cdk.Tags.of(app)``),
265 and ``gco upgrade`` writes cdk.json back byte for byte after its
266 checkout, so both the base deploy and the upgrade's redeploy carry them.
267 """
268 path = clone / "cdk.json"
269 document = json.loads(path.read_text(encoding="utf-8"))
270 settings = document.get("context") if isinstance(document, dict) else None
271 if not isinstance(settings, dict):
272 raise RuntimeError(f"{path} has no context object")
273 tags = settings.get("tags") or {}
274 if not isinstance(tags, dict):
275 raise RuntimeError(f"{path} context.tags must be an object")
276 settings["tags"] = {**tags, run_tag_key: run_id}
277 settings.update(context)
278 path.write_text(json.dumps(document, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
279 return {"path": str(path), "sha256": sha256_file(path), "keys": sorted(context)}
282def read_cloud_assembly(clone: Path) -> dict[str, dict[str, Any]]:
283 """Stack artifacts of the clone's synthesized cloud assembly, by stack name."""
284 manifest = clone / "cdk.out" / "manifest.json"
285 try:
286 document = json.loads(manifest.read_text(encoding="utf-8"))
287 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
288 raise RuntimeError(f"Could not read the base cloud assembly {manifest}: {exc}") from exc
289 artifacts = document.get("artifacts") if isinstance(document, dict) else None
290 if not isinstance(artifacts, dict):
291 raise RuntimeError(f"{manifest} lists no artifacts")
292 stacks: dict[str, dict[str, Any]] = {}
293 for artifact_id, artifact in artifacts.items():
294 if not isinstance(artifact, dict) or artifact.get("type") != "aws:cloudformation:stack":
295 continue
296 properties = artifact.get("properties") or {}
297 name = str(properties.get("stackName") or artifact_id)
298 tags = properties.get("tags")
299 if not isinstance(tags, dict):
300 tags = {
301 str(tag.get("Key")): str(tag.get("Value"))
302 for entries in (artifact.get("metadata") or {}).values()
303 for entry in entries
304 if entry.get("type") == "aws:cdk:stack-tags"
305 for tag in entry.get("data") or []
306 }
307 stacks[name] = {
308 "environment": str(artifact.get("environment") or ""),
309 "tags": {str(key): str(value) for key, value in tags.items()},
310 }
311 return stacks
314def command_environment(
315 workspace: Workspace | None,
316 *,
317 identity: Mapping[str, str] | None = None,
318 base: Mapping[str, str] | None = None,
319) -> dict[str, str]:
320 """The environment every workspace command runs with.
322 ``workspace`` puts the base venv first on ``PATH`` (the CDK app runs
323 ``python3``) and gives the base ``gco`` a cache of its own; ``identity``
324 pins the ``GCO_*`` project and Region settings to the harness's, so a
325 user config file cannot point the base ``gco`` at another deployment.
326 """
327 environment = {
328 key: value
329 for key, value in (os.environ if base is None else base).items()
330 if key not in _SCRUBBED_ENVIRONMENT
331 }
332 if workspace is not None:
333 bin_dir = workspace.venv / "bin"
334 environment["PATH"] = os.pathsep.join(
335 part for part in (str(bin_dir), environment.get("PATH", "")) if part
336 )
337 environment["VIRTUAL_ENV"] = str(workspace.venv)
338 environment["GCO_CACHE_DIR"] = str(workspace.cache)
339 environment["PYTHONNOUSERSITE"] = "1"
340 environment["PIP_DISABLE_PIP_VERSION_CHECK"] = "1"
341 environment.update(identity or {})
342 return environment
345@dataclass(frozen=True)
346class CommandResult:
347 """What one workspace command did."""
349 argv: tuple[str, ...]
350 exit_code: int
351 duration_seconds: float
352 timed_out: bool
353 log_path: Path
354 stdout: str
355 tail: tuple[str, ...]
357 @property
358 def ok(self) -> bool:
359 return self.exit_code == 0 and not self.timed_out
361 def to_dict(self) -> dict[str, Any]:
362 return {
363 "argv": list(self.argv),
364 "exit_code": self.exit_code,
365 "duration_seconds": self.duration_seconds,
366 "timed_out": self.timed_out,
367 "log": str(self.log_path),
368 "tail": list(self.tail),
369 }
372def _terminate(process: subprocess.Popen[str], *, grace_seconds: float) -> None:
373 """Stop the command's whole process group: SIGTERM, then SIGKILL after the grace."""
374 for signum in (signal.SIGTERM, signal.SIGKILL):
375 try:
376 os.killpg(process.pid, signum)
377 except ProcessLookupError:
378 return
379 try:
380 process.wait(timeout=grace_seconds)
381 return
382 except subprocess.TimeoutExpired:
383 continue
386def running_command(pid: int) -> str | None:
387 """The command line of the process with this ID, or ``None`` when there is none."""
388 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed ps argv, no shell
389 ["ps", "-p", str(pid), "-o", "command="],
390 capture_output=True,
391 text=True,
392 check=False,
393 )
394 command = result.stdout.strip()
395 return command if result.returncode == 0 and command else None
398def run_logged(
399 argv: Sequence[str],
400 *,
401 cwd: Path,
402 env: Mapping[str, str],
403 log_path: Path,
404 timeout_seconds: float,
405 echo: Echo | None = None,
406 on_start: Callable[[int], None] | None = None,
407 grace_seconds: float = TERMINATE_GRACE_SECONDS,
408) -> CommandResult:
409 """Run one command, streaming stdout and stderr to ``log_path`` and ``echo``.
411 The command gets its own session, so a timeout, or the harness being
412 interrupted, stops it and every CDK process it started instead of
413 leaving them running. Standard output is also kept whole: ``gco -o json``
414 writes its result document there last. ``on_start`` receives the process
415 ID as soon as the command starts.
416 """
417 tail: deque[str] = deque(maxlen=LOG_TAIL_LINES)
418 stdout_lines: list[str] = []
419 lock = threading.Lock()
420 started = time.monotonic()
421 timed_out = False
422 with log_path.open("w", encoding="utf-8") as log:
423 log.write(f"$ {' '.join(argv)}\n")
424 log.flush()
425 # nosemgrep: dangerous-subprocess-use-audit, python36-compatibility-Popen1, python36-compatibility-Popen2 - fixed argv lists, never a shell; encoding/errors need Python 3.6 and the project requires 3.14
426 process = subprocess.Popen(
427 list(argv),
428 cwd=cwd,
429 env=dict(env),
430 stdout=subprocess.PIPE,
431 stderr=subprocess.PIPE,
432 text=True,
433 encoding="utf-8",
434 errors="replace",
435 start_new_session=True,
436 umask=COMMAND_UMASK,
437 )
439 def pump(stream: IO[str], keep: bool) -> None:
440 for line in stream:
441 with lock:
442 log.write(line)
443 log.flush()
444 tail.append(line.rstrip("\n"))
445 if keep:
446 stdout_lines.append(line)
447 if echo is not None:
448 echo(line)
450 assert process.stdout is not None and process.stderr is not None
451 if on_start is not None:
452 on_start(process.pid)
453 pumps = [
454 threading.Thread(target=pump, args=(process.stdout, True), daemon=True),
455 threading.Thread(target=pump, args=(process.stderr, False), daemon=True),
456 ]
457 for thread in pumps:
458 thread.start()
459 try:
460 try:
461 process.wait(timeout=timeout_seconds)
462 except subprocess.TimeoutExpired:
463 timed_out = True
464 _terminate(process, grace_seconds=grace_seconds)
465 except BaseException:
466 _terminate(process, grace_seconds=grace_seconds)
467 raise
468 finally:
469 for thread in pumps:
470 thread.join()
471 return CommandResult(
472 argv=tuple(argv),
473 exit_code=process.returncode,
474 duration_seconds=round(time.monotonic() - started, 3),
475 timed_out=timed_out,
476 log_path=log_path,
477 stdout="".join(stdout_lines),
478 tail=tuple(tail),
479 )
482def console_echo(prefix: str) -> Echo:
483 """Mirror a command's output on the harness console, one prefixed line at a time."""
485 def echo(line: str) -> None:
486 text = line.rstrip("\n")
487 sys.stdout.write(f"[{prefix}] {text}\n")
488 sys.stdout.flush()
490 return echo
493_DOCUMENT_START = re.compile(r"(?m)^\{")
496def final_json_document(text: str) -> dict[str, Any] | None:
497 """The JSON object ``gco -o json`` writes last, after any streamed CDK output."""
498 for match in reversed(list(_DOCUMENT_START.finditer(text))):
499 try:
500 document = json.loads(text[match.start() :])
501 except ValueError:
502 continue
503 return document if isinstance(document, dict) else None
504 return None
507def gco_result_document(text: str) -> dict[str, Any] | None:
508 """A ``gco -o json`` command's own result document.
510 When the command also printed plain text, ``gco`` wraps everything in
511 ``{"status": "ok", "output": "<text>"}``; the command's document is then
512 the last one inside that text.
513 """
514 document = final_json_document(text)
515 while (
516 document is not None
517 and set(document) == {"status", "output"}
518 and isinstance(document["output"], str)
519 ):
520 inner = final_json_document(document["output"])
521 if inner is None:
522 break
523 document = inner
524 return document