1156 lines
46 KiB
Python
1156 lines
46 KiB
Python
"""G8 release agent fail-closed orchestration tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SCRIPT_PATH = Path(__file__).with_name("run-outcome-os-release-agent.py")
|
|
|
|
|
|
def load_release_agent():
|
|
spec = importlib.util.spec_from_file_location("outcome_os_release_agent", SCRIPT_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("release agent could not be loaded")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@dataclass
|
|
class FakeRunner:
|
|
module: Any
|
|
fail_stage: str | None = None
|
|
builder_shas: tuple[str, str] = ("a" * 64, "a" * 64)
|
|
tracked_status: str = ""
|
|
clean_archive_payloads: tuple[bytes, bytes] = (b"clean archive", b"clean archive")
|
|
clean_head: str = "b" * 40
|
|
clean_tree: str = "c" * 40
|
|
calls: list[str] = field(default_factory=list)
|
|
commands: dict[str, list[str]] = field(default_factory=dict)
|
|
working_directories: dict[str, Path | None] = field(default_factory=dict)
|
|
environments: dict[str, dict[str, str] | None] = field(default_factory=dict)
|
|
timeouts: dict[str, int | None] = field(default_factory=dict)
|
|
event_log: list[str] | None = None
|
|
|
|
def run(self, stage: str, argv: list[str], **kwargs: Any):
|
|
self.calls.append(stage)
|
|
self.commands[stage] = list(argv)
|
|
self.working_directories[stage] = kwargs.get("cwd")
|
|
self.environments[stage] = kwargs.get("env")
|
|
self.timeouts[stage] = kwargs.get("timeout")
|
|
if self.event_log is not None:
|
|
self.event_log.append(f"runner:{stage}")
|
|
if stage == self.fail_stage:
|
|
raise self.module.StageFailure(stage, "synthetic failure")
|
|
if stage.startswith("release_patch_run_"):
|
|
run_index = int(stage.rsplit("_", 1)[1]) - 1
|
|
payload = {
|
|
"ok": True,
|
|
"base_commit": "b" * 40,
|
|
"output_patch": "docs/ops/evidence/release.patch",
|
|
"patch_sha256": self.builder_shas[run_index],
|
|
"patch_bytes": 123,
|
|
"patch_files": 9,
|
|
"git_apply_check": "passed-on-clean-temporary-head",
|
|
"git_apply_cached_check": "passed-on-canonical-clean-index",
|
|
"patch_line_endings": "lf-only",
|
|
"source_worktree": "unchanged-except-output-patch",
|
|
"api_generation": {"sha256": "c" * 64},
|
|
}
|
|
return self.module.CommandResult(0, json.dumps(payload), "")
|
|
if stage == "release_manifest":
|
|
return self.module.CommandResult(0, json.dumps({"ok": True}), "")
|
|
if stage == "clean_head_repository":
|
|
return self.module.CommandResult(0, str(kwargs["cwd"].resolve()) + "\n", "")
|
|
if stage.startswith("clean_head_worktree_"):
|
|
return self.module.CommandResult(0, self.tracked_status, "")
|
|
if stage.startswith("clean_head_head_"):
|
|
return self.module.CommandResult(0, self.clean_head + "\n", "")
|
|
if stage.startswith("clean_head_tree_"):
|
|
return self.module.CommandResult(0, self.clean_tree + "\n", "")
|
|
if stage.startswith("clean_head_archive_run_"):
|
|
run_index = int(stage.rsplit("_", 1)[1]) - 1
|
|
output_index = argv.index("--output")
|
|
Path(argv[output_index + 1]).write_bytes(self.clean_archive_payloads[run_index])
|
|
return self.module.CommandResult(0, "", "")
|
|
return self.module.CommandResult(0, f"{stage}: passed", "")
|
|
|
|
|
|
@dataclass
|
|
class FakeDeployment:
|
|
module: Any
|
|
active_sha: str | None
|
|
rollback_matches: bool = True
|
|
tracked_api_image: str = "sha256:" + "1" * 64
|
|
calls: list[str] = field(default_factory=list)
|
|
event_log: list[str] | None = None
|
|
|
|
def _record(self, event: str) -> None:
|
|
self.calls.append(event)
|
|
if self.event_log is not None:
|
|
self.event_log.append(f"deployment:{event}")
|
|
|
|
def read_active_state(self):
|
|
self._record("read_active_state")
|
|
if self.active_sha is None:
|
|
return None
|
|
return {
|
|
"schema": "vignette.release-agent-active.v1",
|
|
"active_sha": self.active_sha,
|
|
"compose_project": "vignette-preview-20260807",
|
|
"remote_root": "/volume1/docker/vignette-preview-20260807",
|
|
"compose_file": "/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml",
|
|
"env_file": "/volume1/docker/vignette-preview-20260807/infra/.env",
|
|
"api_image": self.tracked_api_image,
|
|
"web_image": "sha256:" + "2" * 64,
|
|
}
|
|
|
|
def snapshot(self):
|
|
self._record("snapshot")
|
|
return self.module.DeploymentSnapshot(
|
|
api_image="sha256:" + "1" * 64,
|
|
web_image="sha256:" + "2" * 64,
|
|
compose_file="/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml",
|
|
env_file="/volume1/docker/vignette-preview-20260807/infra/.env",
|
|
active_sha=self.active_sha,
|
|
)
|
|
|
|
def promote(self, candidate: Path, desired_sha: str, snapshot: Any):
|
|
self._record("promote")
|
|
return {
|
|
"active_sha": desired_sha,
|
|
"api_image": "sha256:" + "3" * 64,
|
|
"web_image": "sha256:" + "4" * 64,
|
|
"candidate": str(candidate),
|
|
"previous_api_image": snapshot.api_image,
|
|
}
|
|
|
|
def commit_active_state(self, state: dict[str, Any]):
|
|
self._record("commit_active_state")
|
|
self.active_sha = state["active_sha"]
|
|
|
|
def rollback(self, snapshot: Any):
|
|
self._record("rollback")
|
|
self.active_sha = snapshot.active_sha
|
|
return {"api_image": snapshot.api_image, "web_image": snapshot.web_image}
|
|
|
|
def verify_rollback(self, snapshot: Any):
|
|
self._record("verify_rollback")
|
|
return {
|
|
"ok": self.rollback_matches,
|
|
"api_image": snapshot.api_image if self.rollback_matches else "sha256:" + "9" * 64,
|
|
"web_image": snapshot.web_image,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class FakeRuntimeProbe:
|
|
module: Any
|
|
fail_on_call: int | None = None
|
|
calls: int = 0
|
|
|
|
def verify(self):
|
|
self.calls += 1
|
|
if self.calls == self.fail_on_call:
|
|
raise self.module.StageFailure("runtime_contract", "synthetic unhealthy runtime")
|
|
return {
|
|
"health": {"status": "ok", "db": True, "engine": True},
|
|
"auth_status": 401,
|
|
"openapi_paths": 122,
|
|
"goals": [f"G{number}" for number in range(9)],
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class TransientRuntimeProbe:
|
|
module: Any
|
|
transient_failures: int
|
|
calls: int = 0
|
|
|
|
def verify(self):
|
|
self.calls += 1
|
|
if self.calls <= self.transient_failures:
|
|
raise self.module.StageFailure(
|
|
"runtime_contract",
|
|
"/api/health status=502, expected=200",
|
|
)
|
|
return {
|
|
"health": {"status": "ok", "db": True, "engine": True},
|
|
"auth_status": 401,
|
|
"openapi_paths": 122,
|
|
"goals": [f"G{number}" for number in range(9)],
|
|
}
|
|
|
|
|
|
class FakeCandidate:
|
|
def __init__(self, root: Path):
|
|
self.root = root
|
|
self.cleaned = False
|
|
self.materialization_mode: str | None = None
|
|
self.archive_sha256: str | None = None
|
|
self.archive_bytes: bytes | None = None
|
|
|
|
def materialize(self, base_commit: str, patch_path: str):
|
|
del base_commit, patch_path
|
|
self.materialization_mode = "patch"
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
return self.root
|
|
|
|
def materialize_archive(self, archive_path: Path, expected_sha256: str):
|
|
self.materialization_mode = "clean-head"
|
|
self.archive_sha256 = expected_sha256
|
|
self.archive_bytes = archive_path.read_bytes()
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
return self.root
|
|
|
|
def cleanup(self):
|
|
self.cleaned = True
|
|
|
|
|
|
@dataclass
|
|
class CaptureRunner:
|
|
module: Any
|
|
calls: list[tuple[str, list[str]]] = field(default_factory=list)
|
|
|
|
def run(self, stage: str, argv: list[str], **_: Any):
|
|
self.calls.append((stage, argv))
|
|
if stage == "nas_promote":
|
|
return self.module.CommandResult(
|
|
0,
|
|
"sha256:" + "3" * 64 + "\n" + "sha256:" + "4" * 64 + "\n",
|
|
"",
|
|
)
|
|
return self.module.CommandResult(0, "", "")
|
|
|
|
|
|
@dataclass
|
|
class ActiveStateRunner:
|
|
module: Any
|
|
payload: dict[str, Any]
|
|
calls: list[str] = field(default_factory=list)
|
|
|
|
def run(self, stage: str, argv: list[str], **_: Any):
|
|
del argv
|
|
self.calls.append(stage)
|
|
return self.module.CommandResult(0, json.dumps(self.payload), "")
|
|
|
|
|
|
@dataclass
|
|
class CandidateArchiveRunner:
|
|
module: Any
|
|
calls: list[tuple[str, list[str]]] = field(default_factory=list)
|
|
|
|
def run(self, stage: str, argv: list[str], **_: Any):
|
|
self.calls.append((stage, argv))
|
|
if stage == "candidate_archive":
|
|
output_index = argv.index("--output")
|
|
archive_path = Path(argv[output_index + 1])
|
|
with tarfile.open(archive_path, "w") as archive:
|
|
marker = archive_path.parent / "README.md"
|
|
marker.write_bytes(b"candidate\r\n")
|
|
archive.add(marker, arcname="README.md")
|
|
generated = archive_path.parent / "api.gen.ts"
|
|
generated.write_bytes(b"export type A = 1;\r\nexport type B = 2;\r\n")
|
|
archive.add(generated, arcname="apps/web/src/lib/api.gen.ts")
|
|
shell_script = archive_path.parent / "99_app_role.sh"
|
|
shell_script.write_bytes(b"#!/usr/bin/env bash\r\necho ready\r\n")
|
|
archive.add(shell_script, arcname="infra/db/init/99_app_role.sh")
|
|
return self.module.CommandResult(0, "", "")
|
|
|
|
|
|
class OutcomeReleaseAgentTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.agent_module = load_release_agent()
|
|
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
manifest = {
|
|
"classifications": {
|
|
"related": [
|
|
{
|
|
"disposition": "candidate-whole-file",
|
|
"paths": list(self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS),
|
|
}
|
|
]
|
|
},
|
|
"release_assembly": {
|
|
"patch_sha256": "a" * 64,
|
|
"output_patch": "docs/ops/evidence/release.patch",
|
|
}
|
|
}
|
|
manifest_path = self.root / "docs/ops/manifest.json"
|
|
manifest_path.parent.mkdir(parents=True)
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
self.manifest_path = manifest_path
|
|
self.milestone = self.agent_module.MaterialMilestone(
|
|
milestone_id="g8-release-agent-close-loop",
|
|
material=True,
|
|
goals=("G8",),
|
|
evidence_refs=("tests:g8-release-agent",),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp.cleanup()
|
|
|
|
def make_agent(
|
|
self,
|
|
*,
|
|
active_sha: str | None,
|
|
execute: bool,
|
|
fail_stage: str | None = None,
|
|
builder_shas: tuple[str, str] = ("a" * 64, "a" * 64),
|
|
source_mode: str = "patch",
|
|
tracked_status: str = "",
|
|
clean_archive_payloads: tuple[bytes, bytes] = (
|
|
b"clean archive",
|
|
b"clean archive",
|
|
),
|
|
source_repo_root: Path | None = None,
|
|
runtime_fail_on_call: int | None = None,
|
|
rollback_matches: bool = True,
|
|
):
|
|
event_log: list[str] = []
|
|
runner = FakeRunner(
|
|
module=self.agent_module,
|
|
fail_stage=fail_stage,
|
|
builder_shas=builder_shas,
|
|
tracked_status=tracked_status,
|
|
clean_archive_payloads=clean_archive_payloads,
|
|
event_log=event_log,
|
|
)
|
|
deployment = FakeDeployment(
|
|
self.agent_module,
|
|
active_sha,
|
|
rollback_matches,
|
|
event_log=event_log,
|
|
)
|
|
runtime = FakeRuntimeProbe(self.agent_module, runtime_fail_on_call)
|
|
candidate = FakeCandidate(self.root / "candidate")
|
|
config = self.agent_module.ReleaseAgentConfig(
|
|
repo_root=source_repo_root or self.root,
|
|
manifest_path=self.manifest_path,
|
|
hunk_map_path=self.root / "docs/ops/hunk-map.json",
|
|
source_mode=source_mode,
|
|
execute=execute,
|
|
milestone=self.milestone,
|
|
target=self.agent_module.NAS_PREVIEW_TARGET,
|
|
evidence_out=self.root / "evidence.json",
|
|
nas_env_file=self.root / "infra/.env.nas-preview",
|
|
)
|
|
agent = self.agent_module.ReleaseAgent(
|
|
config,
|
|
runner=runner,
|
|
deployment=deployment,
|
|
runtime_probe=runtime,
|
|
candidate_manager=candidate,
|
|
)
|
|
return agent, runner, deployment, runtime, candidate
|
|
|
|
def test_rejects_any_target_outside_the_isolated_preview(self) -> None:
|
|
wrong = self.agent_module.DeploymentTarget(
|
|
base_url="http://100.116.83.60:8080",
|
|
compose_project="vignette",
|
|
remote_root="/volume1/docker/vignette",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "isolated NAS preview"):
|
|
self.agent_module.validate_target(wrong)
|
|
|
|
def test_mobile_shell_and_late_migrations_are_required_release_payloads(self) -> None:
|
|
manifest = {
|
|
"classifications": {
|
|
"related": [
|
|
{
|
|
"disposition": "candidate-whole-file",
|
|
"paths": [
|
|
path
|
|
for path in self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS
|
|
if path != "apps/web/src/components/shell/shell.css"
|
|
],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
with self.assertRaisesRegex(
|
|
self.agent_module.StageFailure,
|
|
"apps/web/src/components/shell/shell.css",
|
|
):
|
|
self.agent_module.validate_required_release_payload(manifest)
|
|
|
|
def test_admin_usage_index_migration_is_idempotent_and_release_required(self) -> None:
|
|
migration_path = (
|
|
SCRIPT_PATH.parent.parent
|
|
/ "infra"
|
|
/ "db"
|
|
/ "init"
|
|
/ "18_admin_usage_ledger_index.sql"
|
|
)
|
|
migration = migration_path.read_text(encoding="utf-8")
|
|
|
|
self.assertNotIn("BEGIN;", migration)
|
|
self.assertIn(
|
|
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_turns_admin_usage_created_at",
|
|
migration,
|
|
)
|
|
self.assertIn("ON app.turns (created_at DESC)", migration)
|
|
self.assertIn("WHERE speaker = 'client';", migration)
|
|
self.assertNotIn("COMMIT;", migration)
|
|
self.assertIn(
|
|
"18_admin_usage_ledger_index.sql",
|
|
self.agent_module.RELEASE_DB_MIGRATIONS,
|
|
)
|
|
self.assertIn(
|
|
"18_admin_usage_ledger_index.sql",
|
|
self.agent_module.RELEASE_DB_ONLINE_MIGRATIONS,
|
|
)
|
|
self.assertIn(
|
|
"infra/db/init/18_admin_usage_ledger_index.sql",
|
|
self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS,
|
|
)
|
|
|
|
def test_two_patch_runs_must_be_byte_identical_before_any_deploy(self) -> None:
|
|
agent, runner, deployment, _, _ = self.make_agent(
|
|
active_sha=None,
|
|
execute=True,
|
|
builder_shas=("a" * 64, "d" * 64),
|
|
)
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
self.assertEqual("release_determinism", raised.exception.report["failed_stage"])
|
|
self.assertNotIn("snapshot", deployment.calls)
|
|
self.assertNotIn("promote", deployment.calls)
|
|
self.assertEqual(
|
|
["release_patch_run_1", "release_patch_run_2"],
|
|
runner.calls,
|
|
)
|
|
|
|
def test_parser_keeps_patch_and_controller_repo_defaults(self) -> None:
|
|
parser = self.agent_module.build_parser()
|
|
default_args = parser.parse_args(["--milestone", "milestone.json"])
|
|
detached_root = self.root / "detached"
|
|
clean_args = parser.parse_args(
|
|
[
|
|
"--milestone",
|
|
"milestone.json",
|
|
"--source-mode",
|
|
"clean-head",
|
|
"--source-repo-root",
|
|
str(detached_root),
|
|
]
|
|
)
|
|
|
|
self.assertEqual("patch", default_args.source_mode)
|
|
self.assertEqual(self.agent_module.REPO_ROOT, default_args.source_repo_root)
|
|
self.assertEqual("clean-head", clean_args.source_mode)
|
|
self.assertEqual(detached_root, clean_args.source_repo_root)
|
|
|
|
def test_source_repo_root_must_be_an_absolute_existing_directory(self) -> None:
|
|
relative = Path("detached-source")
|
|
missing = self.root / "missing"
|
|
source_file = self.root / "source-file"
|
|
source_file.write_text("not a directory\n", encoding="utf-8")
|
|
|
|
for source_root, message in (
|
|
(relative, "must be absolute"),
|
|
(missing, "does not exist"),
|
|
(source_file, "not a directory"),
|
|
):
|
|
with self.subTest(source_root=source_root):
|
|
with self.assertRaisesRegex(self.agent_module.StageFailure, message):
|
|
self.agent_module.validate_source_repo_root(source_root)
|
|
|
|
def test_clean_head_override_runs_every_git_command_in_source_repo(self) -> None:
|
|
detached_root = self.root / "detached-source"
|
|
detached_root.mkdir()
|
|
archive_payload = b"detached clean-head archive"
|
|
agent, runner, _, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
source_mode="clean-head",
|
|
clean_archive_payloads=(archive_payload, archive_payload),
|
|
source_repo_root=detached_root,
|
|
)
|
|
|
|
report = agent.run()
|
|
|
|
git_stages = [
|
|
stage
|
|
for stage in runner.calls
|
|
if stage == "clean_head_repository"
|
|
or stage.startswith("clean_head_worktree_")
|
|
or stage.startswith("clean_head_head_")
|
|
or stage.startswith("clean_head_tree_")
|
|
or stage.startswith("clean_head_archive_run_")
|
|
]
|
|
self.assertEqual(9, len(git_stages))
|
|
self.assertEqual(
|
|
{detached_root},
|
|
{runner.working_directories[stage] for stage in git_stages},
|
|
)
|
|
self.assertEqual(str(detached_root), report["source"]["repo_root"])
|
|
self.assertEqual("clean-head", candidate.materialization_mode)
|
|
|
|
def test_clean_head_rejects_tracked_changes_and_ignores_untracked_in_query(self) -> None:
|
|
agent, runner, deployment, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
source_mode="clean-head",
|
|
tracked_status=" M docs/TODO.md\n",
|
|
)
|
|
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
|
|
self.assertEqual("clean_head_worktree", raised.exception.report["failed_stage"])
|
|
self.assertEqual(
|
|
["clean_head_repository", "clean_head_worktree_before"],
|
|
runner.calls,
|
|
)
|
|
self.assertIn(
|
|
"--untracked-files=no",
|
|
runner.commands["clean_head_worktree_before"],
|
|
)
|
|
self.assertEqual([], deployment.calls)
|
|
self.assertIsNone(candidate.materialization_mode)
|
|
|
|
def test_clean_head_non_git_source_fails_before_identity_or_deployment(self) -> None:
|
|
agent, runner, deployment, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
source_mode="clean-head",
|
|
fail_stage="clean_head_repository",
|
|
)
|
|
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
|
|
self.assertEqual("clean_head_repository", raised.exception.report["failed_stage"])
|
|
self.assertEqual(["clean_head_repository"], runner.calls)
|
|
self.assertEqual([], deployment.calls)
|
|
self.assertIsNone(candidate.materialization_mode)
|
|
|
|
def test_clean_head_archive_runs_must_be_byte_identical(self) -> None:
|
|
agent, runner, deployment, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
source_mode="clean-head",
|
|
clean_archive_payloads=(b"first archive", b"second archive"),
|
|
)
|
|
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
|
|
self.assertEqual("clean_head_determinism", raised.exception.report["failed_stage"])
|
|
self.assertEqual(2, len([stage for stage in runner.calls if "archive_run" in stage]))
|
|
self.assertEqual([], deployment.calls)
|
|
self.assertIsNone(candidate.materialization_mode)
|
|
|
|
def test_clean_head_evidence_binds_exact_head_tree_and_archive(self) -> None:
|
|
archive_payload = b"deterministic clean-head archive"
|
|
expected_sha = self.agent_module.sha256_bytes(archive_payload)
|
|
agent, runner, deployment, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
source_mode="clean-head",
|
|
clean_archive_payloads=(archive_payload, archive_payload),
|
|
)
|
|
|
|
report = agent.run()
|
|
|
|
self.assertTrue(report["ok"])
|
|
self.assertEqual(expected_sha, report["desired_sha"])
|
|
self.assertEqual("clean-head", report["source_mode"])
|
|
self.assertEqual("b" * 40, report["source"]["head"])
|
|
self.assertEqual("c" * 40, report["source"]["tree"])
|
|
self.assertEqual(expected_sha, report["source"]["archive"]["sha256"])
|
|
self.assertEqual(len(archive_payload), report["source"]["archive"]["bytes"])
|
|
self.assertEqual(2, report["source"]["archive"]["deterministic_runs"])
|
|
self.assertEqual("clean-head", candidate.materialization_mode)
|
|
self.assertEqual(expected_sha, candidate.archive_sha256)
|
|
self.assertEqual(archive_payload, candidate.archive_bytes)
|
|
self.assertNotIn("release_manifest", runner.calls)
|
|
self.assertFalse(any(stage.startswith("release_patch_run_") for stage in runner.calls))
|
|
self.assertIn("promote", deployment.calls)
|
|
evidence = json.loads((self.root / "evidence.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(report["source"], evidence["source"])
|
|
self.assertEqual("clean-head", evidence["source_mode"])
|
|
|
|
def test_same_deployment_sha_is_a_runtime_checked_noop(self) -> None:
|
|
agent, runner, deployment, runtime, candidate = self.make_agent(
|
|
active_sha="a" * 64,
|
|
execute=True,
|
|
)
|
|
report = agent.run()
|
|
self.assertTrue(report["ok"])
|
|
self.assertEqual("noop_same_sha", report["status"])
|
|
self.assertEqual(1, runtime.calls)
|
|
self.assertIn("snapshot", deployment.calls)
|
|
self.assertNotIn("promote", deployment.calls)
|
|
self.assertNotIn("api_tests", runner.calls)
|
|
self.assertFalse(candidate.cleaned)
|
|
|
|
def test_same_sha_runtime_image_drift_is_not_adopted_as_a_noop(self) -> None:
|
|
agent, _, deployment, runtime, _ = self.make_agent(
|
|
active_sha="a" * 64,
|
|
execute=True,
|
|
)
|
|
deployment.tracked_api_image = "sha256:" + "9" * 64
|
|
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
|
|
self.assertEqual("deployment_state", raised.exception.report["failed_stage"])
|
|
self.assertEqual(0, runtime.calls)
|
|
self.assertNotIn("promote", deployment.calls)
|
|
|
|
def test_default_dry_run_reports_changed_sha_without_mutation(self) -> None:
|
|
agent, runner, deployment, runtime, _ = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=False,
|
|
)
|
|
report = agent.run()
|
|
self.assertTrue(report["ok"])
|
|
self.assertEqual("dry_run_changed_sha", report["status"])
|
|
self.assertEqual("a" * 64, report["desired_sha"])
|
|
self.assertEqual("e" * 64, report["active_sha"])
|
|
self.assertEqual(0, runtime.calls)
|
|
self.assertNotIn("snapshot", deployment.calls)
|
|
self.assertNotIn("promote", deployment.calls)
|
|
self.assertNotIn("api_tests", runner.calls)
|
|
|
|
def test_untracked_remote_state_blocks_execute_before_snapshot(self) -> None:
|
|
agent, runner, deployment, _, _ = self.make_agent(
|
|
active_sha=None,
|
|
execute=True,
|
|
)
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
self.assertEqual("deployment_state", raised.exception.report["failed_stage"])
|
|
self.assertNotIn("snapshot", deployment.calls)
|
|
self.assertNotIn("api_tests", runner.calls)
|
|
|
|
def test_changed_sha_runs_all_gates_then_promotion_and_browser_review(self) -> None:
|
|
agent, runner, deployment, runtime, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
)
|
|
report = agent.run()
|
|
self.assertTrue(report["ok"])
|
|
self.assertEqual("deployed_ssot_sync_required", report["status"])
|
|
self.assertEqual("patch", report["source_mode"])
|
|
self.assertEqual("a" * 64, report["source"]["patch"]["sha256"])
|
|
self.assertEqual("a" * 64, report["active_sha"])
|
|
self.assertEqual("e" * 64, report["previous_sha"])
|
|
self.assertEqual(
|
|
[
|
|
"release_patch_run_1",
|
|
"release_patch_run_2",
|
|
"release_manifest",
|
|
"nas_preflight",
|
|
"api_tests",
|
|
"web_install",
|
|
"web_api_contract",
|
|
"web_typecheck",
|
|
"web_build",
|
|
"source_insecure_context_e2e",
|
|
"session_e2e",
|
|
"postdeploy_browser_review",
|
|
],
|
|
runner.calls,
|
|
)
|
|
self.assertEqual(
|
|
["read_active_state", "snapshot", "promote", "commit_active_state"],
|
|
deployment.calls,
|
|
)
|
|
self.assertEqual(1, runtime.calls)
|
|
self.assertTrue(candidate.cleaned)
|
|
postdeploy_command = runner.commands["postdeploy_browser_review"]
|
|
test_index = postdeploy_command.index("test")
|
|
first_flag = next(
|
|
index
|
|
for index in range(test_index + 1, len(postdeploy_command))
|
|
if postdeploy_command[index].startswith("--")
|
|
)
|
|
self.assertEqual(
|
|
self.agent_module.POSTDEPLOY_NAS_E2E_SPECS,
|
|
tuple(postdeploy_command[test_index + 1 : first_flag]),
|
|
)
|
|
self.assertNotIn("--grep", postdeploy_command)
|
|
self.assertEqual(
|
|
{f"--project={project}" for project in self.agent_module.RELEASE_BROWSER_PROJECTS},
|
|
{value for value in postdeploy_command if value.startswith("--project=")},
|
|
)
|
|
self.assertEqual(
|
|
self.agent_module.RELEASE_E2E_TIMEOUT_SECONDS,
|
|
runner.timeouts["postdeploy_browser_review"],
|
|
)
|
|
self.assertEqual(
|
|
self.agent_module.NAS_PREVIEW_TARGET.base_url,
|
|
runner.environments["postdeploy_browser_review"]["PLAYWRIGHT_BASE_URL"],
|
|
)
|
|
self.assertEqual(
|
|
"1",
|
|
runner.environments["postdeploy_browser_review"]["PLAYWRIGHT_SKIP_WEB_SERVER"],
|
|
)
|
|
self.assertIn(
|
|
"e2e/self-directed-learning-loop.spec.ts",
|
|
self.agent_module.POSTDEPLOY_NAS_E2E_SPECS,
|
|
)
|
|
self.assertIn(
|
|
"e2e/deliberate-practice.spec.ts",
|
|
self.agent_module.POSTDEPLOY_NAS_E2E_SPECS,
|
|
)
|
|
self.assertNotIn(
|
|
"e2e/insecure-context-uuid.spec.ts",
|
|
self.agent_module.POSTDEPLOY_NAS_E2E_SPECS,
|
|
)
|
|
self.assertNotIn(
|
|
"e2e/returned-practice-db-closed-loop.spec.ts",
|
|
self.agent_module.POSTDEPLOY_NAS_E2E_SPECS,
|
|
)
|
|
self.assertEqual(
|
|
set(self.agent_module.RELEASE_UI_E2E_SPECS),
|
|
set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS)
|
|
| set(self.agent_module.POSTDEPLOY_SOURCE_ONLY_E2E_SPECS),
|
|
)
|
|
self.assertEqual(
|
|
set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS),
|
|
set(self.agent_module.POSTDEPLOY_NAS_REAL_API_E2E_SPECS)
|
|
| set(self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS),
|
|
)
|
|
self.assertFalse(
|
|
set(self.agent_module.POSTDEPLOY_NAS_REAL_API_E2E_SPECS)
|
|
& set(self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS)
|
|
)
|
|
self.assertLess(
|
|
runner.event_log.index("runner:postdeploy_browser_review"),
|
|
runner.event_log.index("deployment:commit_active_state"),
|
|
)
|
|
self.assertTrue(
|
|
set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS).issubset(
|
|
set(runner.commands["session_e2e"])
|
|
)
|
|
)
|
|
self.assertNotIn(
|
|
"e2e/insecure-context-uuid.spec.ts",
|
|
runner.commands["session_e2e"],
|
|
)
|
|
source_command = runner.commands["source_insecure_context_e2e"]
|
|
self.assertIn("e2e/insecure-context-uuid.spec.ts", source_command)
|
|
self.assertIn("--project=chromium-desktop", source_command)
|
|
self.assertIn("--project=chromium-mobile", source_command)
|
|
source_environment = runner.environments["source_insecure_context_e2e"]
|
|
self.assertNotIn("PLAYWRIGHT_BASE_URL", source_environment)
|
|
self.assertNotIn("PLAYWRIGHT_SKIP_WEB_SERVER", source_environment)
|
|
self.assertEqual("127.0.0.1", source_environment["PLAYWRIGHT_HOST"])
|
|
self.assertEqual(
|
|
str(self.agent_module.SOURCE_ONLY_E2E_PORT),
|
|
source_environment["PLAYWRIGHT_PORT"],
|
|
)
|
|
self.assertEqual("1", source_environment["CI"])
|
|
self.assertEqual(15 * 60, self.agent_module.RELEASE_E2E_TIMEOUT_SECONDS)
|
|
self.assertIn(
|
|
"e2e/insecure-context-uuid.spec.ts",
|
|
self.agent_module.RELEASE_UI_E2E_SPECS,
|
|
)
|
|
self.assertEqual("patch", candidate.materialization_mode)
|
|
self.assertEqual(
|
|
(
|
|
"14_continuous_improvement.sql",
|
|
"15_self_directed_practice_runtime.sql",
|
|
"16_calibration_transfer_actual_execution.sql",
|
|
"17_improvement_workbook_contracts.sql",
|
|
"18_admin_usage_ledger_index.sql",
|
|
"19_auth_identity_alias.sql",
|
|
),
|
|
self.agent_module.RELEASE_DB_MIGRATIONS,
|
|
)
|
|
evidence = json.loads((self.root / "evidence.json").read_text(encoding="utf-8"))
|
|
self.assertEqual("a" * 64, evidence["active_sha"])
|
|
self.assertEqual("e" * 64, evidence["previous_sha"])
|
|
self.assertEqual("a" * 64, evidence["deployment"]["desired_sha"])
|
|
self.assertEqual("required", evidence["ssot_sync"]["status"])
|
|
self.assertEqual(
|
|
list(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS),
|
|
evidence["browser_review_proof"]["specs"],
|
|
)
|
|
self.assertEqual(
|
|
["e2e/insecure-context-uuid.spec.ts"],
|
|
evidence["browser_review_proof"]["source_only_candidate_specs"],
|
|
)
|
|
self.assertEqual(
|
|
["e2e/returned-practice-db-closed-loop.spec.ts"],
|
|
evidence["browser_review_proof"]["separate_disposable_db_specs"],
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
"real_api_db_specs": [
|
|
"e2e/session-layout.spec.ts",
|
|
"e2e/session-persistence.spec.ts",
|
|
],
|
|
"route_fixture_specs": list(
|
|
self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS
|
|
),
|
|
},
|
|
evidence["browser_review_proof"]["runtime_scope"],
|
|
)
|
|
self.assertEqual(
|
|
"not_run_by_release_agent",
|
|
evidence["browser_review_proof"]["separate_disposable_db_status"],
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
"docs/dev_dashboard.html",
|
|
"docs/TODO.md",
|
|
"docs/ops/backlog-2026-06-26.md",
|
|
"docs/ops/outcome-os-clean-release-manifest-2026-08-07.json",
|
|
],
|
|
evidence["ssot_sync"]["required_paths"],
|
|
)
|
|
|
|
def test_postdeploy_failure_rolls_back_and_proves_previous_images(self) -> None:
|
|
agent, _, deployment, runtime, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
runtime_fail_on_call=1,
|
|
)
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
report = raised.exception.report
|
|
self.assertEqual("runtime_contract", report["failed_stage"])
|
|
self.assertEqual("verified", report["rollback"]["status"])
|
|
self.assertEqual(
|
|
[
|
|
"read_active_state",
|
|
"snapshot",
|
|
"promote",
|
|
"rollback",
|
|
"verify_rollback",
|
|
],
|
|
deployment.calls,
|
|
)
|
|
self.assertEqual(2, runtime.calls)
|
|
self.assertTrue(candidate.cleaned)
|
|
|
|
def test_rollback_image_mismatch_remains_a_hard_failure(self) -> None:
|
|
agent, runner, deployment, _, _ = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
fail_stage="postdeploy_browser_review",
|
|
rollback_matches=False,
|
|
)
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
self.assertEqual("failed", raised.exception.report["rollback"]["status"])
|
|
self.assertIn("rollback image proof mismatch", raised.exception.report["rollback"]["error"])
|
|
self.assertIn("verify_rollback", deployment.calls)
|
|
self.assertNotIn("commit_active_state", deployment.calls)
|
|
self.assertLess(
|
|
runner.event_log.index("runner:postdeploy_browser_review"),
|
|
runner.event_log.index("deployment:rollback"),
|
|
)
|
|
|
|
def test_postpromotion_runtime_retries_only_transient_readiness_failures(self) -> None:
|
|
agent, _, deployment, _, candidate = self.make_agent(
|
|
active_sha="e" * 64,
|
|
execute=True,
|
|
)
|
|
runtime = TransientRuntimeProbe(self.agent_module, transient_failures=2)
|
|
agent.runtime_probe = runtime
|
|
agent.config.runtime_ready_timeout_seconds = 1
|
|
agent.config.runtime_retry_interval_seconds = 0
|
|
|
|
report = agent.run()
|
|
|
|
self.assertTrue(report["ok"])
|
|
self.assertEqual(3, runtime.calls)
|
|
self.assertEqual(3, report["runtime"]["readiness_attempts"])
|
|
self.assertIn("promote", deployment.calls)
|
|
self.assertNotIn("rollback", deployment.calls)
|
|
self.assertTrue(candidate.cleaned)
|
|
|
|
def test_non_material_milestone_fails_before_release_commands(self) -> None:
|
|
agent, runner, deployment, _, _ = self.make_agent(
|
|
active_sha=None,
|
|
execute=True,
|
|
)
|
|
agent.config.milestone = self.agent_module.MaterialMilestone(
|
|
milestone_id="cosmetic-copy-only",
|
|
material=False,
|
|
goals=("G8",),
|
|
evidence_refs=("test:copy",),
|
|
)
|
|
with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised:
|
|
agent.run()
|
|
self.assertEqual("material_milestone", raised.exception.report["failed_stage"])
|
|
self.assertEqual([], runner.calls)
|
|
self.assertEqual([], deployment.calls)
|
|
|
|
def test_remote_rollback_recreates_only_api_web_and_proxy(self) -> None:
|
|
runner = CaptureRunner(self.agent_module)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
snapshot = self.agent_module.DeploymentSnapshot(
|
|
api_image="sha256:" + "1" * 64,
|
|
web_image="sha256:" + "2" * 64,
|
|
compose_file=(
|
|
"/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml"
|
|
),
|
|
env_file="/volume1/docker/vignette-preview-20260807/infra/.env",
|
|
active_sha="e" * 64,
|
|
)
|
|
driver.rollback(snapshot)
|
|
remote_call = next(argv for stage, argv in runner.calls if stage == "nas_rollback")
|
|
remote_command = remote_call[-1]
|
|
self.assertIn("up -d --no-build --no-deps --force-recreate api web proxy", remote_command)
|
|
self.assertNotIn(" compose down", remote_command)
|
|
self.assertNotIn("docker volume", remote_command)
|
|
self.assertNotIn("docker network", remote_command)
|
|
self.assertNotIn("--volumes", remote_command)
|
|
self.assertIn(
|
|
"--env-file /volume1/docker/vignette-preview-20260807/infra/.env",
|
|
remote_command,
|
|
)
|
|
|
|
def test_remote_promotion_reuses_snapshot_env_before_upload(self) -> None:
|
|
root = "/volume1/docker/vignette-preview-20260807"
|
|
runner = CaptureRunner(self.agent_module)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
candidate = self.root / "candidate"
|
|
candidate.mkdir()
|
|
(candidate / "README.md").write_text("candidate\n", encoding="utf-8")
|
|
snapshot = self.agent_module.DeploymentSnapshot(
|
|
api_image="sha256:" + "1" * 64,
|
|
web_image="sha256:" + "2" * 64,
|
|
compose_file=f"{root}/release/infra/docker-compose.yml",
|
|
env_file=f"{root}/release/infra/.env",
|
|
active_sha="e" * 64,
|
|
)
|
|
|
|
state = driver.promote(candidate, "a" * 64, snapshot)
|
|
|
|
stages = [stage for stage, _ in runner.calls]
|
|
self.assertLess(stages.index("nas_prepare_release"), stages.index("nas_upload_archive"))
|
|
prepare_command = next(
|
|
argv[-1] for stage, argv in runner.calls if stage == "nas_prepare_release"
|
|
)
|
|
promote_command = next(
|
|
argv[-1] for stage, argv in runner.calls if stage == "nas_promote"
|
|
)
|
|
self.assertIn("required env file missing", prepare_command)
|
|
self.assertIn(f"--env-file {root}/release/infra/.env", promote_command)
|
|
self.assertNotIn(f"--env-file {root}/infra/.env", promote_command)
|
|
transactional_migration_command = next(
|
|
line
|
|
for line in promote_command.splitlines()
|
|
if "17_improvement_workbook_contracts.sql" in line
|
|
and "psql -v ON_ERROR_STOP=1" in line
|
|
)
|
|
online_migration_command = next(
|
|
line
|
|
for line in promote_command.splitlines()
|
|
if "18_admin_usage_ledger_index.sql" in line
|
|
and "psql -v ON_ERROR_STOP=1" in line
|
|
)
|
|
self.assertIn("--single-transaction", transactional_migration_command)
|
|
self.assertNotIn("--single-transaction", online_migration_command)
|
|
self.assertEqual(f"{root}/release/infra/.env", state["env_file"])
|
|
|
|
def test_adopted_legacy_state_infers_compose_adjacent_env_file(self) -> None:
|
|
root = "/volume1/docker/vignette-preview-20260807"
|
|
runner = ActiveStateRunner(
|
|
self.agent_module,
|
|
{
|
|
"schema": "vignette.release-agent-active.v1",
|
|
"active_sha": "e" * 64,
|
|
"compose_project": "vignette-preview-20260807",
|
|
"remote_root": root,
|
|
"compose_file": f"{root}/release/infra/docker-compose.yml",
|
|
"api_image": "sha256:" + "1" * 64,
|
|
"web_image": "sha256:" + "2" * 64,
|
|
},
|
|
)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
|
|
state = driver.read_active_state()
|
|
|
|
self.assertEqual(f"{root}/release/infra/.env", state["env_file"])
|
|
|
|
def test_remote_copy_uses_synology_compatible_legacy_scp_protocol(self) -> None:
|
|
runner = CaptureRunner(self.agent_module)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
source = self.root / "state.json"
|
|
source.write_text("{}\n", encoding="utf-8")
|
|
|
|
driver._scp(
|
|
"nas_upload_test",
|
|
source,
|
|
"/volume1/docker/vignette-preview-20260807/incoming/state.json",
|
|
)
|
|
|
|
argv = next(argv for stage, argv in runner.calls if stage == "nas_upload_test")
|
|
self.assertEqual("scp.exe", argv[0])
|
|
self.assertIn("-O", argv)
|
|
|
|
def test_candidate_normalizes_generated_contract_and_shell_scripts_to_lf(self) -> None:
|
|
runner = CandidateArchiveRunner(self.agent_module)
|
|
manager = self.agent_module.CleanCandidateManager(self.root, runner)
|
|
|
|
candidate = manager.materialize("b" * 40, "release.patch")
|
|
|
|
argv = next(argv for stage, argv in runner.calls if stage == "candidate_patch_apply")
|
|
self.assertEqual(
|
|
["git", "apply", "--whitespace=nowarn"],
|
|
argv[:-1],
|
|
)
|
|
self.assertEqual(
|
|
b"export type A = 1;\nexport type B = 2;\n",
|
|
(candidate / "apps/web/src/lib/api.gen.ts").read_bytes(),
|
|
)
|
|
self.assertEqual(
|
|
b"#!/usr/bin/env bash\necho ready\n",
|
|
(candidate / "infra/db/init/99_app_role.sh").read_bytes(),
|
|
)
|
|
self.assertEqual(b"candidate\r\n", (candidate / "README.md").read_bytes())
|
|
manager.cleanup()
|
|
|
|
def test_clean_head_candidate_is_archive_only_with_platform_normalization(self) -> None:
|
|
runner = CandidateArchiveRunner(self.agent_module)
|
|
manager = self.agent_module.CleanCandidateManager(self.root, runner)
|
|
archive_path = self.root / "clean-head.tar"
|
|
source = self.root / "source.txt"
|
|
source.write_bytes(b"exact committed bytes\r\n")
|
|
generated_api = self.root / "api.gen.ts"
|
|
generated_api.write_bytes(b"export type Exact = true;\r\n")
|
|
shell_script = self.root / "99_app_role.sh"
|
|
shell_script.write_bytes(b"#!/usr/bin/env bash\r\necho ready\r\n")
|
|
with tarfile.open(archive_path, "w") as archive:
|
|
archive.add(source, arcname="source.txt")
|
|
archive.add(generated_api, arcname="apps/web/src/lib/api.gen.ts")
|
|
archive.add(shell_script, arcname="infra/db/init/99_app_role.sh")
|
|
expected_sha = self.agent_module.sha256_bytes(archive_path.read_bytes())
|
|
|
|
candidate = manager.materialize_archive(archive_path, expected_sha)
|
|
|
|
self.assertEqual(b"exact committed bytes\r\n", (candidate / "source.txt").read_bytes())
|
|
self.assertEqual(
|
|
b"export type Exact = true;\n",
|
|
(candidate / "apps/web/src/lib/api.gen.ts").read_bytes(),
|
|
)
|
|
self.assertEqual(
|
|
b"#!/usr/bin/env bash\necho ready\n",
|
|
(candidate / "infra/db/init/99_app_role.sh").read_bytes(),
|
|
)
|
|
self.assertEqual([], runner.calls)
|
|
manager.cleanup()
|
|
|
|
def test_clean_head_candidate_rejects_archive_sha_drift(self) -> None:
|
|
runner = CandidateArchiveRunner(self.agent_module)
|
|
manager = self.agent_module.CleanCandidateManager(self.root, runner)
|
|
archive_path = self.root / "clean-head.tar"
|
|
archive_path.write_bytes(b"not the bound archive")
|
|
|
|
with self.assertRaisesRegex(
|
|
self.agent_module.StageFailure,
|
|
"archive SHA drifted",
|
|
):
|
|
manager.materialize_archive(archive_path, "a" * 64)
|
|
|
|
self.assertEqual([], runner.calls)
|
|
|
|
def test_active_state_rejects_noncanonical_compose_path_before_snapshot(self) -> None:
|
|
root = "/volume1/docker/vignette-preview-20260807"
|
|
runner = ActiveStateRunner(
|
|
self.agent_module,
|
|
{
|
|
"schema": "vignette.release-agent-active.v1",
|
|
"active_sha": "e" * 64,
|
|
"compose_project": "vignette-preview-20260807",
|
|
"remote_root": root,
|
|
"compose_file": f"{root}/releases/../other/infra/docker-compose.yml",
|
|
"api_image": "sha256:" + "1" * 64,
|
|
"web_image": "sha256:" + "2" * 64,
|
|
},
|
|
)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
|
|
with self.assertRaisesRegex(self.agent_module.StageFailure, "Compose file"):
|
|
driver.read_active_state()
|
|
|
|
self.assertEqual(["nas_read_active_state"], runner.calls)
|
|
|
|
def test_active_state_accepts_only_the_exact_audited_legacy_release_path(self) -> None:
|
|
root = "/volume1/docker/vignette-preview-20260807"
|
|
payload = {
|
|
"schema": "vignette.release-agent-active.v1",
|
|
"active_sha": "e" * 64,
|
|
"compose_project": "vignette-preview-20260807",
|
|
"remote_root": root,
|
|
"compose_file": f"{root}/release/infra/docker-compose.yml",
|
|
"api_image": "sha256:" + "1" * 64,
|
|
"web_image": "sha256:" + "2" * 64,
|
|
}
|
|
runner = ActiveStateRunner(self.agent_module, payload)
|
|
driver = self.agent_module.NasPreviewDeploymentDriver(
|
|
runner,
|
|
ssh_target="release-agent@100.116.83.60",
|
|
)
|
|
|
|
state = driver.read_active_state()
|
|
self.assertEqual(payload, {key: state[key] for key in payload})
|
|
self.assertEqual(f"{root}/release/infra/.env", state["env_file"])
|
|
self.assertEqual(["nas_read_active_state"], runner.calls)
|
|
|
|
def test_candidate_archive_excludes_release_agent_secret_env(self) -> None:
|
|
candidate = self.root / "archive-candidate"
|
|
(candidate / ".release-agent").mkdir(parents=True)
|
|
(candidate / ".release-agent/candidate.env").write_text(
|
|
"OPENAI_API_KEY=must-not-ship\n",
|
|
encoding="utf-8",
|
|
)
|
|
(candidate / "README.md").write_text("safe\n", encoding="utf-8")
|
|
destination = self.root / "candidate.tar.gz"
|
|
|
|
self.agent_module._archive_candidate(candidate, destination)
|
|
|
|
with tarfile.open(destination, "r:gz") as archive:
|
|
names = archive.getnames()
|
|
self.assertIn("README.md", names)
|
|
self.assertFalse(any(name == ".release-agent" or name.startswith(".release-agent/") for name in names))
|
|
|
|
def test_candidate_compose_project_is_unique_per_process(self) -> None:
|
|
desired_sha = "a" * 64
|
|
first = self.agent_module.candidate_compose_project(desired_sha, process_id=101)
|
|
second = self.agent_module.candidate_compose_project(desired_sha, process_id=202)
|
|
|
|
self.assertNotEqual(first, second)
|
|
self.assertEqual("vignette-release-gate-aaaaaaaaaaaa-p101", first)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|