From 65cdcdd457cd5c12396c8d9b8633708a2e450861 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 3 Jun 2026 09:39:25 +0100 Subject: [PATCH] Add v5 agent self-update runtime proof Extend the unified-agent RC rehearsal with an optional local v5-to-v6 process-swap check. Record the v5.1.34 to v6.0.0-rc.6 proof result for the upgrade-continuity gate. --- ...-agent-v5-upgrade-continuity-2026-03-12.md | 11 + .../internal/unified_agent_rc_rehearsal.py | 524 ++++++++++++++++-- .../unified_agent_rc_rehearsal_test.py | 190 +++++++ 3 files changed, 675 insertions(+), 50 deletions(-) diff --git a/docs/release-control/v6/internal/records/unified-agent-v5-upgrade-continuity-2026-03-12.md b/docs/release-control/v6/internal/records/unified-agent-v5-upgrade-continuity-2026-03-12.md index d2cba56bd..f3d48a124 100644 --- a/docs/release-control/v6/internal/records/unified-agent-v5-upgrade-continuity-2026-03-12.md +++ b/docs/release-control/v6/internal/records/unified-agent-v5-upgrade-continuity-2026-03-12.md @@ -14,6 +14,17 @@ - `python3 scripts/release_control/unified_agent_rc_rehearsal.py --base-url http://127.0.0.1:7655 --expected-version 6.0.0-rc.1 --release-base-url file:///tmp/pulse-private-release-assets --arch linux-amd64 --api-token --expected-active-agents 3 --expected-agent-name delly --expected-online-agents 3` - Result: pass +## Automated Local Runtime Proof + +- Date: `2026-06-03` +- Host architecture: `darwin-arm64` +- Starting binary: `v5.1.34` built from `repos/pulse-5.1.x` tag `v5.1.34` +- Target binary: `v6.0.0-rc.6` built from `pulse/v6-release` +- Command: `PYTHONPATH=scripts/release_control/internal python3 scripts/release_control/internal/unified_agent_rc_rehearsal.py --base-url http://127.0.0.1:1 --expected-version 6.0.0-rc.6 --release-base-url http://127.0.0.1:1/releases --skip-asset-checks --runtime-v5-agent /tmp/pulse-v5-v6-runtime-proof/bin/pulse-agent-v5.1.34-darwin-arm64 --runtime-v6-agent /tmp/pulse-v5-v6-runtime-proof/bin/pulse-agent-v6.0.0-rc.6-darwin-arm64 --runtime-expected-from v5.1.34 --runtime-expected-to v6.0.0-rc.6 --runtime-timeout 90 --json` +- Result: pass +- Observed result: the v5 process reported `v5.1.34`, downloaded v6 binary checksum `407d803270388e36ddc6c851c0dd04bdb2ff9e4d7c08cda9d4cf5839df790227`, exec'd into `v6.0.0-rc.6`, and reported `updated_from=v5.1.34`. +- Scope note: this local proof covers the in-place process swap, checksum-gated download, v5 host-report compatibility, v6 agent-report compatibility, and first-report `updated_from`. Live active-agent accounting still needs the authenticated Pulse API rehearsal above when proving a full server environment. + ## Manual Crossover Exercise 1. Built a real `linux-amd64` v5 agent from `main` with version `5.1.23`. diff --git a/scripts/release_control/internal/unified_agent_rc_rehearsal.py b/scripts/release_control/internal/unified_agent_rc_rehearsal.py index 5f7a564f9..cfbdf928b 100644 --- a/scripts/release_control/internal/unified_agent_rc_rehearsal.py +++ b/scripts/release_control/internal/unified_agent_rc_rehearsal.py @@ -4,9 +4,17 @@ from __future__ import annotations import argparse +import gzip import hashlib import json +import os +import shutil +import subprocess +import tempfile +import threading +import time from dataclasses import asdict, dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Iterable from urllib import error, parse, request @@ -22,6 +30,104 @@ class CheckResult: detail: str +class _RuntimeProofHandler(BaseHTTPRequestHandler): + server: "_RuntimeProofServer" + + def do_GET(self) -> None: # noqa: N802 + self.server.record_request("GET", self.path) + parsed = parse.urlparse(self.path) + path = parsed.path + + if path == "/api/agent/version": + self._write_json({"version": self.server.expected_version}) + return + + if path == "/download/pulse-agent": + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("X-Checksum-Sha256", self.server.v6_checksum) + self.send_header("Content-Length", str(len(self.server.v6_binary))) + self.end_headers() + self.wfile.write(self.server.v6_binary) + return + + if path in {"/api/agents/host/lookup", "/api/agents/agent/lookup"}: + if path.endswith("/host/lookup"): + self._write_json({"success": True, "host": {"id": self.server.agent_id}}) + else: + self._write_json({"success": True, "agent": {"id": self.server.agent_id}}) + return + + if path.startswith("/api/agents/host/") and path.endswith("/config"): + self._write_json({"success": True, "hostId": self.server.agent_id, "config": {}}) + return + + if path.startswith("/api/agents/agent/") and path.endswith("/config"): + self._write_json({"success": True, "agentId": self.server.agent_id, "config": {}}) + return + + self.send_response(404) + self.end_headers() + self.wfile.write(b"missing") + + def do_POST(self) -> None: # noqa: N802 + self.server.record_request("POST", self.path) + body = self.rfile.read(int(self.headers.get("Content-Length", "0") or "0")) + if self.headers.get("Content-Encoding", "").lower() == "gzip": + body = gzip.decompress(body) + + if self.path in {"/api/agents/host/report", "/api/agents/agent/report"}: + try: + payload = json.loads(body.decode("utf-8")) + except json.JSONDecodeError: + self.send_response(400) + self.end_headers() + self.wfile.write(b"bad report json") + return + self.server.record_report(self.path, payload) + if self.path.endswith("/agent/report"): + self._write_json({"success": True, "agentId": self.server.agent_id, "config": {}}) + else: + self._write_json({"success": True, "hostId": self.server.agent_id, "config": {}}) + return + + self.send_response(404) + self.end_headers() + self.wfile.write(b"missing") + + def log_message(self, format: str, *args: object) -> None: # noqa: A003 + return + + def _write_json(self, payload: dict[str, object]) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +class _RuntimeProofServer(ThreadingHTTPServer): + def __init__(self, *, expected_version: str, v6_binary: bytes, agent_id: str) -> None: + super().__init__(("127.0.0.1", 0), _RuntimeProofHandler) + self.expected_version = expected_version + self.v6_binary = v6_binary + self.v6_checksum = sha256_hex(v6_binary) + self.agent_id = agent_id + self.requests: list[tuple[str, str]] = [] + self.reports: list[tuple[str, dict[str, object]]] = [] + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self.server_port}" + + def record_request(self, method: str, path: str) -> None: + self.requests.append((method, path)) + + def record_report(self, path: str, payload: dict[str, object]) -> None: + self.reports.append((path, payload)) + + def safe_check(name: str, fn) -> CheckResult: try: return fn() @@ -118,6 +224,46 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: default="Unified Agent RC Rehearsal", help="Markdown title used when writing --report-out", ) + parser.add_argument( + "--skip-asset-checks", + action="store_true", + help=( + "Skip live Pulse version/install/download asset checks; intended for isolated " + "local runtime self-update proof runs" + ), + ) + parser.add_argument( + "--runtime-v5-agent", + help="Optional path to a real v5 pulse-agent binary to exercise in-place self-update", + ) + parser.add_argument( + "--runtime-v6-agent", + help="Optional path to the v6 pulse-agent binary served by the local runtime proof server", + ) + parser.add_argument( + "--runtime-expected-from", + default="v5.1.34", + help="Expected version reported by --runtime-v5-agent before the swap", + ) + parser.add_argument( + "--runtime-expected-to", + help="Expected version after the swap; defaults to --expected-version", + ) + parser.add_argument( + "--runtime-timeout", + type=float, + default=60.0, + help="Seconds to wait for the real v5 process to self-update and report as v6", + ) + parser.add_argument( + "--runtime-work-dir", + help="Optional existing directory for runtime proof artifacts; defaults to a temporary directory", + ) + parser.add_argument( + "--runtime-keep-work-dir", + action="store_true", + help="Keep runtime proof artifacts instead of deleting the temporary directory", + ) return parser.parse_args(argv) @@ -182,6 +328,35 @@ def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() +def normalize_version(value: str) -> str: + value = value.strip() + if value.startswith("v"): + value = value[1:] + return value + + +def report_agent_info(report: dict[str, object]) -> dict[str, object]: + agent = report.get("agent") + if isinstance(agent, dict): + return agent + return {} + + +def agent_report_version(report: dict[str, object]) -> str: + agent = report_agent_info(report) + return str(agent.get("version", "")).strip() + + +def agent_report_updated_from(report: dict[str, object]) -> str: + agent = report_agent_info(report) + return str(agent.get("updatedFrom") or agent.get("updated_from") or "").strip() + + +def agent_report_type(report: dict[str, object]) -> str: + agent = report_agent_info(report) + return str(agent.get("type", "")).strip() + + def release_asset_url(release_base_url: str, version: str, asset_name: str) -> str: return f"{normalize_base_url(release_base_url)}/v{version}/{asset_name}" @@ -275,6 +450,180 @@ def check_update_info(update_info_dir: str, expected_updated_from: str) -> Check ) +def check_local_runtime_self_update( + *, + v5_agent: str, + v6_agent: str, + expected_from: str, + expected_to: str, + timeout: float, + work_dir: str | None, + keep_work_dir: bool, +) -> CheckResult: + v5_path = Path(v5_agent) + v6_path = Path(v6_agent) + if not v5_path.is_file(): + return CheckResult("local-runtime-self-update", False, f"v5 agent not found: {v5_path}") + if not v6_path.is_file(): + return CheckResult("local-runtime-self-update", False, f"v6 agent not found: {v6_path}") + + expected_from_output = subprocess.check_output([str(v5_path), "--version"], text=True).strip() + if normalize_version(expected_from_output) != normalize_version(expected_from): + return CheckResult( + "local-runtime-self-update", + False, + f"v5 binary reports {expected_from_output!r}, expected {expected_from!r}", + ) + + expected_to_output = subprocess.check_output([str(v6_path), "--version"], text=True).strip() + if normalize_version(expected_to_output) != normalize_version(expected_to): + return CheckResult( + "local-runtime-self-update", + False, + f"v6 binary reports {expected_to_output!r}, expected {expected_to!r}", + ) + + v6_binary = v6_path.read_bytes() + agent_id = "agent-v5-to-v6-runtime-proof" + server = _RuntimeProofServer( + expected_version=normalize_version(expected_to), + v6_binary=v6_binary, + agent_id=agent_id, + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + temp_root_created = False + if work_dir: + root = Path(work_dir) + root.mkdir(parents=True, exist_ok=True) + else: + root = Path(tempfile.mkdtemp(prefix="pulse-agent-v5-v6-proof-")) + temp_root_created = True + + proc: subprocess.Popen[str] | None = None + try: + runtime_agent = root / "pulse-agent" + shutil.copy2(v5_path, runtime_agent) + runtime_agent.chmod(0o755) + token_file = root / "token" + token_file.write_text("runtime-proof-token\n", encoding="utf-8") + token_file.chmod(0o600) + agent_id_file = root / "agent-id" + state_dir = root / "state" + state_dir.mkdir(exist_ok=True) + log_path = root / "agent.log" + + env = os.environ.copy() + env.update( + { + "PULSE_AGENT_CONFIG_SIGNATURE_REQUIRED": "false", + "PULSE_STATE_DIR": str(state_dir), + } + ) + with log_path.open("w", encoding="utf-8") as log_file: + proc = subprocess.Popen( + [ + str(runtime_agent), + "--url", + server.base_url, + "--token-file", + str(token_file), + "--agent-id", + agent_id, + "--agent-id-file", + str(agent_id_file), + "--hostname", + "pulse-v5-v6-runtime-proof", + "--interval", + "2s", + "--health-addr", + "127.0.0.1:0", + "--insecure", + "--log-level", + "debug", + ], + cwd=str(root), + env=env, + stdout=log_file, + stderr=subprocess.STDOUT, + text=True, + ) + + deadline = time.monotonic() + timeout + saw_v5 = False + saw_v6 = False + saw_updated_from = False + while time.monotonic() < deadline: + if proc.poll() is not None: + break + for _path, report in server.reports: + version = normalize_version(agent_report_version(report)) + if version == normalize_version(expected_from): + saw_v5 = True + if version == normalize_version(expected_to): + saw_v6 = True + if normalize_version(agent_report_updated_from(report)) == normalize_version( + expected_from + ): + saw_updated_from = True + if agent_report_type(report) != "unified": + return CheckResult( + "local-runtime-self-update", + False, + f"v6 report type={agent_report_type(report)!r}, expected 'unified'", + ) + if saw_v5 and saw_v6 and saw_updated_from: + replaced_version = subprocess.check_output( + [str(runtime_agent), "--version"], text=True + ).strip() + if normalize_version(replaced_version) != normalize_version(expected_to): + return CheckResult( + "local-runtime-self-update", + False, + f"replaced binary reports {replaced_version!r}, expected {expected_to!r}", + ) + return CheckResult( + "local-runtime-self-update", + True, + ( + f"v5 process reported {expected_from_output}, downloaded checksum " + f"{server.v6_checksum}, exec'd v6 report {expected_to_output} with " + f"updated_from={expected_from_output}; work_dir={root}" + ), + ) + time.sleep(0.25) + + log_excerpt = "" + if log_path.exists(): + log_lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines() + log_excerpt = "; log_tail=" + " | ".join(log_lines[-8:]) + versions = [agent_report_version(report) for _path, report in server.reports] + return CheckResult( + "local-runtime-self-update", + False, + ( + f"timed out waiting for v5->v6 report sequence " + f"(saw_v5={saw_v5}, saw_v6={saw_v6}, " + f"saw_updated_from={saw_updated_from}, versions={versions!r}, " + f"work_dir={root}){log_excerpt}" + ), + ) + finally: + if proc is not None and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + server.shutdown() + server.server_close() + thread.join(timeout=5) + if temp_root_created and not keep_work_dir: + shutil.rmtree(root, ignore_errors=True) + + def check_active_agent_accounting( *, base_url: str, @@ -433,44 +782,64 @@ def run_rehearsal(args: argparse.Namespace) -> list[CheckResult]: base_url = normalize_base_url(args.base_url) release_base_url = normalize_base_url(args.release_base_url) auth_headers = build_auth_headers(args) - results = [ - safe_check( - "agent-version-endpoint", - lambda: check_version(base_url, args.expected_version, args.timeout), - ), - safe_check( - "install-sh-asset", - lambda: compare_asset( - name="install-sh-asset", - live_url=f"{base_url}/install.sh", - release_url=release_asset_url(release_base_url, args.expected_version, "install.sh"), - timeout=args.timeout, - ), - ), - safe_check( - "install-ps1-asset", - lambda: compare_asset( - name="install-ps1-asset", - live_url=f"{base_url}/install.ps1", - release_url=release_asset_url(release_base_url, args.expected_version, "install.ps1"), - timeout=args.timeout, - ), - ), - ] + results: list[CheckResult] = [] - for arch in args.arch: - quoted_arch = parse.quote(arch, safe="") + if not args.skip_asset_checks: + results.extend( + [ + safe_check( + "agent-version-endpoint", + lambda: check_version(base_url, args.expected_version, args.timeout), + ), + safe_check( + "install-sh-asset", + lambda: compare_asset( + name="install-sh-asset", + live_url=f"{base_url}/install.sh", + release_url=release_asset_url( + release_base_url, args.expected_version, "install.sh" + ), + timeout=args.timeout, + ), + ), + safe_check( + "install-ps1-asset", + lambda: compare_asset( + name="install-ps1-asset", + live_url=f"{base_url}/install.ps1", + release_url=release_asset_url( + release_base_url, args.expected_version, "install.ps1" + ), + timeout=args.timeout, + ), + ), + ] + ) + + for arch in args.arch: + quoted_arch = parse.quote(arch, safe="") + results.append( + safe_check( + f"agent-binary-{arch}", + lambda arch=arch, quoted_arch=quoted_arch: compare_asset( + name=f"agent-binary-{arch}", + live_url=f"{base_url}/download/pulse-agent?arch={quoted_arch}", + release_url=release_asset_url( + release_base_url, args.expected_version, agent_binary_asset_name(arch) + ), + timeout=args.timeout, + expect_checksum_header=True, + ), + ) + ) + elif args.arch: results.append( safe_check( - f"agent-binary-{arch}", - lambda arch=arch, quoted_arch=quoted_arch: compare_asset( - name=f"agent-binary-{arch}", - live_url=f"{base_url}/download/pulse-agent?arch={quoted_arch}", - release_url=release_asset_url( - release_base_url, args.expected_version, agent_binary_asset_name(arch) - ), - timeout=args.timeout, - expect_checksum_header=True, + "asset-check-selection", + lambda: CheckResult( + name="asset-check-selection", + ok=False, + detail="--arch cannot be combined with --skip-asset-checks", ), ) ) @@ -519,19 +888,60 @@ def run_rehearsal(args: argparse.Namespace) -> list[CheckResult]: ) ) + if args.runtime_v5_agent or args.runtime_v6_agent: + if not args.runtime_v5_agent or not args.runtime_v6_agent: + results.append( + CheckResult( + name="local-runtime-self-update", + ok=False, + detail="both --runtime-v5-agent and --runtime-v6-agent are required together", + ) + ) + else: + results.append( + safe_check( + "local-runtime-self-update", + lambda: check_local_runtime_self_update( + v5_agent=args.runtime_v5_agent, + v6_agent=args.runtime_v6_agent, + expected_from=args.runtime_expected_from, + expected_to=args.runtime_expected_to or args.expected_version, + timeout=args.runtime_timeout, + work_dir=args.runtime_work_dir, + keep_work_dir=args.runtime_keep_work_dir, + ), + ) + ) + + if not results: + results.append( + CheckResult( + name="check-selection", + ok=False, + detail="no rehearsal checks selected", + ) + ) + return results def render_text(results: Iterable[CheckResult]) -> str: + results = list(results) lines: list[str] = [] for result in results: prefix = "PASS" if result.ok else "FAIL" lines.append(f"[{prefix}] {result.name}: {result.detail}") - lines.append( - "Manual follow-up still required: confirm the upgraded v5-installed agent " - "reconnects as one canonical v6 identity, surfaces `updated_from` exactly once, " - "and leaves user-visible active-agent counts aligned with runtime enforcement." - ) + if any(result.name == "local-runtime-self-update" and result.ok for result in results): + lines.append( + "Runtime follow-up covered: local v5 pulse-agent performed an in-place " + "self-update, exec'd the v6 binary, and reported `updated_from` once." + ) + else: + lines.append( + "Manual follow-up still required: confirm the upgraded v5-installed agent " + "reconnects as one canonical v6 identity, surfaces `updated_from` exactly once, " + "and leaves user-visible active-agent counts aligned with runtime enforcement." + ) return "\n".join(lines) @@ -543,6 +953,7 @@ def render_markdown_report( release_base_url: str, results: Iterable[CheckResult], ) -> str: + results = list(results) lines = [ f"# {title}", "", @@ -556,16 +967,29 @@ def render_markdown_report( for result in results: status = "PASS" if result.ok else "FAIL" lines.append(f"- `{status}` `{result.name}`: {result.detail}") - lines.extend( - [ - "", - "## Manual Follow-up", - "", - "- Confirm the upgraded v5-installed agent reconnects as one canonical v6 identity.", - "- Confirm `updated_from` appears exactly once on the first canonical v6 report and clears on the next report.", - "- Confirm settings/billing active-agent counts still match runtime enforcement after the upgrade.", - ] - ) + runtime_covered = any(result.name == "local-runtime-self-update" and result.ok for result in results) + if runtime_covered: + lines.extend( + [ + "", + "## Runtime Proof", + "", + "- Local v5 `pulse-agent` performed the in-place self-update and exec'd the v6 binary.", + "- The first v6 report carried `updated_from` for the v5 source version.", + "- Active-agent accounting still requires a live Pulse API check when this proof is run outside a full server.", + ] + ) + else: + lines.extend( + [ + "", + "## Manual Follow-up", + "", + "- Confirm the upgraded v5-installed agent reconnects as one canonical v6 identity.", + "- Confirm `updated_from` appears exactly once on the first canonical v6 report and clears on the next report.", + "- Confirm settings/billing active-agent counts still match runtime enforcement after the upgrade.", + ] + ) return "\n".join(lines) diff --git a/scripts/release_control/internal/unified_agent_rc_rehearsal_test.py b/scripts/release_control/internal/unified_agent_rc_rehearsal_test.py index 649ea3092..bb80faf8c 100644 --- a/scripts/release_control/internal/unified_agent_rc_rehearsal_test.py +++ b/scripts/release_control/internal/unified_agent_rc_rehearsal_test.py @@ -3,6 +3,7 @@ from __future__ import annotations import contextlib import io import json +import shutil import tempfile import threading import unittest @@ -12,9 +13,11 @@ from pathlib import Path from unified_agent_rc_rehearsal import ( agent_binary_asset_name, + check_local_runtime_self_update, check_update_info, main, render_markdown_report, + render_text, release_asset_url, run_rehearsal, summarize_http_error_body, @@ -61,6 +64,101 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase): _FixtureHandler.routes = routes _FixtureHandler.request_headers = {} + def write_fake_runtime_agents(self, root: Path) -> tuple[Path, Path]: + v5 = root / "fake-v5-agent" + v6 = root / "fake-v6-agent" + v5.write_text( + """#!/usr/bin/env python3 +import json +import os +import sys +import urllib.request + +def arg(name): + if name in sys.argv: + index = sys.argv.index(name) + if index + 1 < len(sys.argv): + return sys.argv[index + 1] + return "" + +if "--version" in sys.argv: + print("v5.1.34") + sys.exit(0) + +base_url = arg("--url").rstrip("/") +agent_id = arg("--agent-id") or "agent-v5" + +def post(path, payload): + request = urllib.request.Request( + base_url + path, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + urllib.request.urlopen(request, timeout=5).read() + +post("/api/agents/host/report", {"agent": {"id": agent_id, "type": "unified", "version": "v5.1.34"}}) +urllib.request.urlopen(base_url + "/api/agent/version", timeout=5).read() +download = urllib.request.urlopen(base_url + "/download/pulse-agent?arch=darwin-arm64", timeout=5).read() +with open(sys.argv[0], "wb") as handle: + handle.write(download) +os.chmod(sys.argv[0], 0o755) +with open(os.path.join(os.path.dirname(sys.argv[0]), ".pulse-update-info"), "w", encoding="utf-8") as handle: + handle.write("v5.1.34") +os.execv(sys.argv[0], sys.argv) +""", + encoding="utf-8", + ) + v6.write_text( + """#!/usr/bin/env python3 +import json +import os +import sys +import time +import urllib.request + +def arg(name): + if name in sys.argv: + index = sys.argv.index(name) + if index + 1 < len(sys.argv): + return sys.argv[index + 1] + return "" + +if "--version" in sys.argv: + print("v6.0.0-rc.6") + sys.exit(0) + +base_url = arg("--url").rstrip("/") +agent_id = arg("--agent-id") or "agent-v6" +info_path = os.path.join(os.path.dirname(sys.argv[0]), ".pulse-update-info") +updated_from = "" +if os.path.exists(info_path): + with open(info_path, encoding="utf-8") as handle: + updated_from = handle.read().strip() + os.remove(info_path) +payload = { + "agent": { + "id": agent_id, + "type": "unified", + "version": "v6.0.0-rc.6", + "updatedFrom": updated_from, + } +} +request = urllib.request.Request( + base_url + "/api/agents/agent/report", + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", +) +urllib.request.urlopen(request, timeout=5).read() +time.sleep(60) +""", + encoding="utf-8", + ) + v5.chmod(0o755) + v6.chmod(0o755) + return v5, v6 + def test_release_asset_url(self) -> None: got = release_asset_url("https://example.invalid/releases/download/", "6.0.0-rc.1", "install.sh") self.assertEqual(got, "https://example.invalid/releases/download/v6.0.0-rc.1/install.sh") @@ -113,6 +211,14 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase): "expected_agent_name": [], "expected_online_agents": None, "json": False, + "skip_asset_checks": False, + "runtime_v5_agent": None, + "runtime_v6_agent": None, + "runtime_expected_from": "v5.1.34", + "runtime_expected_to": None, + "runtime_timeout": 60.0, + "runtime_work_dir": None, + "runtime_keep_work_dir": False, }, )() results = run_rehearsal(args) @@ -217,6 +323,90 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase): self.assertIn("`FAIL` `agent-binary-linux-amd64`", report) self.assertIn("## Manual Follow-up", report) + def test_render_text_accepts_one_pass_iterable(self) -> None: + result = type( + "R", + (), + { + "name": "local-runtime-self-update", + "ok": True, + "detail": "runtime proof passed", + }, + )() + rendered = render_text(item for item in [result]) + self.assertIn("[PASS] local-runtime-self-update", rendered) + self.assertIn("Runtime follow-up covered", rendered) + + def test_main_refuses_empty_runtime_only_selection(self) -> None: + exit_code = run_main( + [ + "--base-url", + f"{self.base_url}/pulse", + "--expected-version", + "6.0.0-rc.1", + "--release-base-url", + f"{self.base_url}/releases", + "--skip-asset-checks", + "--json", + ] + ) + self.assertEqual(exit_code, 1) + + def test_main_refuses_arch_when_asset_checks_are_skipped(self) -> None: + exit_code = run_main( + [ + "--base-url", + f"{self.base_url}/pulse", + "--expected-version", + "6.0.0-rc.1", + "--release-base-url", + f"{self.base_url}/releases", + "--skip-asset-checks", + "--arch", + "linux-amd64", + "--json", + ] + ) + self.assertEqual(exit_code, 1) + + def test_local_runtime_self_update_exercises_process_swap(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + v5, v6 = self.write_fake_runtime_agents(Path(tmp)) + + result = check_local_runtime_self_update( + v5_agent=str(v5), + v6_agent=str(v6), + expected_from="v5.1.34", + expected_to="v6.0.0-rc.6", + timeout=10, + work_dir=None, + keep_work_dir=False, + ) + + self.assertTrue(result.ok, result.detail) + + def test_local_runtime_self_update_keeps_temp_work_dir_when_requested(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + v5, v6 = self.write_fake_runtime_agents(Path(tmp)) + + result = check_local_runtime_self_update( + v5_agent=str(v5), + v6_agent=str(v6), + expected_from="v5.1.34", + expected_to="v6.0.0-rc.6", + timeout=10, + work_dir=None, + keep_work_dir=True, + ) + + self.assertTrue(result.ok, result.detail) + work_dir = Path(result.detail.rsplit("work_dir=", maxsplit=1)[1]) + try: + self.assertTrue(work_dir.exists()) + self.assertTrue((work_dir / "pulse-agent").exists()) + finally: + shutil.rmtree(work_dir, ignore_errors=True) + def test_main_writes_report(self) -> None: install = b"#!/bin/sh\necho install\n" ps1 = b"Write-Output 'install'\n"