Agent-System/scripts/tests/test_agentctl_w4.py

#!/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.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")


if __name__ == "__main__":
    unittest.main()