WeiCHEN
33c7b9a260
Load configs/agent-governance.json from the main worktree with fail-closed POLICY_* codes, and keep test fixtures writing the same contract file.
189 lines
8.0 KiB
Python
189 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Black-box tests for agentctl_w4 projection apply and deploy/recover loop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import shutil
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
W4_SOURCE = SOURCE / "agentctl_w4.py"
|
|
|
|
|
|
def load_w4(path: Path):
|
|
spec = importlib.util.spec_from_file_location("agentctl_w4_under_test", path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
class AgentctlW4Tests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="agentctl-w4-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / "repo"
|
|
self.root.mkdir()
|
|
(self.root / "scripts").mkdir()
|
|
shutil.copy2(W4_SOURCE, self.root / "scripts" / "agentctl_w4.py")
|
|
(self.root / "configs").mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(SOURCE.parent / "configs" / "agent-governance.json",
|
|
self.root / "configs" / "agent-governance.json")
|
|
self.w4 = load_w4(self.root / "scripts" / "agentctl_w4.py")
|
|
self.db_path = Path(self.tmp.name) / "w4.sqlite"
|
|
self.db = sqlite3.connect(self.db_path)
|
|
self.db.row_factory = sqlite3.Row
|
|
self.addCleanup(self.db.close)
|
|
self.w4.ensure_schema(self.db)
|
|
|
|
def _write_current(self, body: str) -> Path:
|
|
path = self.root / ".doc" / "ops" / "CURRENT.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(body, encoding="utf-8")
|
|
return path
|
|
|
|
def test_row_projection_cas_and_insert(self):
|
|
doc = (
|
|
"# Coordination\n\n"
|
|
"| Task | Owner | Branch | State |\n"
|
|
"| --- | --- | --- | --- |\n"
|
|
"| alpha | Alice | feature/alpha | active |\n"
|
|
)
|
|
target = self._write_current(doc)
|
|
current_row = "| alpha | Alice | feature/alpha | active |"
|
|
replacement = "| alpha | Alice | feature/alpha | done |"
|
|
change = self.w4.build_row_projection_change(
|
|
file=".doc/ops/CURRENT.md",
|
|
record_id="alpha",
|
|
current_row=current_row,
|
|
replacement=replacement,
|
|
)
|
|
result = self.w4.apply_projection_locally(self.root, change)
|
|
self.assertEqual(result["status"], "applied")
|
|
self.assertIn(replacement, target.read_text(encoding="utf-8"))
|
|
self.assertNotIn(current_row, target.read_text(encoding="utf-8"))
|
|
|
|
insert_row = "| gamma | Carol | feature/gamma | active |"
|
|
insert = self.w4.build_row_projection_change(
|
|
file=".doc/ops/CURRENT.md",
|
|
record_id="gamma",
|
|
current_row=None,
|
|
replacement=insert_row,
|
|
)
|
|
self.assertIsNone(insert["expected_sha256"])
|
|
inserted = self.w4.apply_projection_locally(self.root, insert)
|
|
self.assertEqual(inserted["status"], "applied")
|
|
text = target.read_text(encoding="utf-8")
|
|
self.assertIn(insert_row, text)
|
|
self.assertIn(replacement, text)
|
|
|
|
def test_document_projection_cas_mismatch_refuses(self):
|
|
original = "# Features\n\nplanned\n"
|
|
path = self.root / ".doc" / "status" / "FEATURE-MATRIX.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(original, encoding="utf-8")
|
|
change = self.w4.build_document_projection_change(
|
|
file=".doc/status/FEATURE-MATRIX.md",
|
|
current_text=original,
|
|
replacement="# Features\n\ndone\n",
|
|
)
|
|
# Mutate on disk so live hash no longer matches the CAS tip.
|
|
path.write_text("# Features\n\nchanged-underneath\n", encoding="utf-8")
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
self.w4.apply_projection_locally(self.root, change)
|
|
self.assertIn("BASE_CHANGED", str(ctx.exception))
|
|
self.assertEqual(path.read_text(encoding="utf-8"), "# Features\n\nchanged-underneath\n")
|
|
|
|
def test_deploy_recover_loop_success_with_fake_command(self):
|
|
outcome = self.w4.execute_deploy_recover_loop(
|
|
self.db,
|
|
deploy_id="dep-ok",
|
|
task="modular-decouple-residual-2026-09-23",
|
|
env="dev",
|
|
expected_commit="abcdef1",
|
|
command=["/bin/echo", "ok"],
|
|
)
|
|
self.assertEqual(outcome["status"], "passed")
|
|
self.assertEqual(outcome["deploy"]["phase"], "passed")
|
|
self.assertTrue(outcome["receipt"].get("ok"))
|
|
self.assertTrue(outcome["receipt"].get("accepted"))
|
|
stored = self.w4.deployment_get(self.db, "dep-ok")
|
|
self.assertEqual(stored["phase"], "passed")
|
|
receipt = json.loads(stored["receipt"])
|
|
self.assertTrue(receipt.get("accepted"))
|
|
|
|
def test_deploy_recover_loop_failure_sets_recovery_required(self):
|
|
outcome = self.w4.execute_deploy_recover_loop(
|
|
self.db,
|
|
deploy_id="dep-fail",
|
|
task="modular-decouple-residual-2026-09-23",
|
|
env="dev",
|
|
expected_commit="abcdef2",
|
|
command=["/bin/false"],
|
|
)
|
|
self.assertEqual(outcome["status"], "recovery_required")
|
|
self.assertEqual(outcome["deploy"]["phase"], "recovery_required")
|
|
self.assertFalse(outcome["receipt"].get("ok"))
|
|
stored = self.w4.deployment_get(self.db, "dep-fail")
|
|
self.assertEqual(stored["phase"], "recovery_required")
|
|
pending = self.w4.queue_list(self.db, task="modular-decouple-residual-2026-09-23",
|
|
state="pending")
|
|
self.assertTrue(any(j["id"] == "dep-fail-recover-pending" for j in pending))
|
|
|
|
def test_recover_deploy_from_recovery_required(self):
|
|
self.w4.execute_deploy_recover_loop(
|
|
self.db,
|
|
deploy_id="dep-recover",
|
|
task="modular-decouple-residual-2026-09-23",
|
|
env="dev",
|
|
expected_commit="abcdef3",
|
|
command=[sys.executable, "-c", "raise SystemExit(1)"],
|
|
)
|
|
recovered = self.w4.recover_deploy(
|
|
self.db, "dep-recover", reason="operator cleared after fake failure")
|
|
self.assertEqual(recovered["phase"], "passed")
|
|
self.assertIsNone(recovered["last_error"])
|
|
receipt = json.loads(recovered["receipt"])
|
|
self.assertTrue(receipt.get("recovered"))
|
|
self.assertEqual(receipt.get("recover_reason"),
|
|
"operator cleared after fake failure")
|
|
|
|
def test_deploy_phase_transition_rejects_illegal_jump(self):
|
|
deploy = self.w4.deployment_create(
|
|
self.db, deploy_id="dep-phase", task="phase-task",
|
|
env="dev", expected_commit="abcdef4", manifest=None, dry_run=True)
|
|
self.assertEqual(deploy["phase"], "prepared")
|
|
self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
|
|
self.w4.deployment_set_phase(self.db, "dep-phase", "accepting")
|
|
passed = self.w4.deployment_set_phase(self.db, "dep-phase", "passed")
|
|
self.assertEqual(passed["phase"], "passed")
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
|
|
self.assertIn("illegal transition passed -> deploying", str(ctx.exception))
|
|
still = self.w4.deployment_get(self.db, "dep-phase")
|
|
self.assertEqual(still["phase"], "passed")
|
|
|
|
|
|
class DarwinCwdVisibilityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.w4 = load_w4(W4_SOURCE)
|
|
|
|
def test_cwd_visibility_structured_on_host(self):
|
|
with tempfile.TemporaryDirectory(prefix="w4-cwd-") as temp:
|
|
report = self.w4.cwd_visibility(Path(temp))
|
|
self.assertIn(report["status"], ("clear", "busy", "unreadable", "unsupported"))
|
|
self.assertIsInstance(report.get("pids"), list)
|
|
if sys.platform == "darwin":
|
|
self.assertNotEqual(report["status"], "unsupported")
|
|
self.assertNotIn("/proc", report.get("message", ""))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|