coord/tests/test_install.py

#!/usr/bin/env python3
"""Behavioral tests for fresh install and legacy in-place migration."""

from __future__ import annotations

import importlib.util
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[1]
INSTALL = ROOT / "install"
RELEASE = (ROOT / "VERSION").read_text(encoding="utf-8").strip()


class InstallBehavior(unittest.TestCase):
    def run_install(
        self, target: Path, *args: str
    ) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            [sys.executable, str(INSTALL), str(target), *args],
            cwd=ROOT,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    def test_fresh_install_requires_explicit_immutable_project_id(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            target = Path(temporary)
            missing = self.run_install(target)
            self.assertEqual(missing.returncode, 2)
            self.assertIn("--project-id", missing.stderr)
            installed = self.run_install(target, "--project-id", "fresh-project")
            self.assertEqual(installed.returncode, 0, installed.stderr)
            self.assertEqual(
                (target / ".coord/project-id").read_text().strip(),
                "fresh-project",
            )
            codex_skill = target / ".agents/skills/coord/SKILL.md"
            claude_skill = target / ".claude/skills/coord/SKILL.md"
            canonical_skill = target / ".coord/skills/coord/SKILL.md"
            self.assertEqual(codex_skill.read_bytes(), claude_skill.read_bytes())
            self.assertEqual(codex_skill.read_bytes(), canonical_skill.read_bytes())
            self.assertEqual(
                codex_skill.read_bytes(),
                (ROOT / "skills/coord/SKILL.md").read_bytes(),
            )
            self.assertEqual(
                (target / ".agents/skills/coord/agents/openai.yaml").read_bytes(),
                (
                    target / ".claude/skills/coord/agents/openai.yaml"
                ).read_bytes(),
            )
            self.assertTrue((target / ".coord/AGENTS.md").is_file())
            self.assertTrue((target / ".coord/CLAUDE.md").is_file())
            self.assertEqual(
                (target / ".coord/VERSION").read_text().strip(), RELEASE
            )
            self.assertFalse((target / "AGENTS.md").exists())
            self.assertFalse((target / "CLAUDE.md").exists())
            checked = self.run_install(target, "--check")
            self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
            self.assertIn(f"matches {RELEASE}", checked.stdout)

    def test_doctor_reports_unversioned_drift_and_repair_is_targeted(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            target = Path(temporary)
            legacy = target / ".claude/coord"
            legacy.mkdir(parents=True)
            (legacy / "project-id").write_text("doctor-project\n")
            stale = target / ".coord/coord"
            stale.parent.mkdir(parents=True)
            stale.write_text("stale managed file\n")
            unrelated = target / ".agents/skills/local/SKILL.md"
            unrelated.parent.mkdir(parents=True)
            unrelated.write_text("operator content\n")

            diagnosed = self.run_install(target, "--doctor")
            self.assertEqual(diagnosed.returncode, 1, diagnosed.stderr)
            self.assertIn(f"source version {RELEASE}", diagnosed.stdout)
            self.assertIn("installed version unversioned", diagnosed.stdout)
            self.assertIn("drift", diagnosed.stdout)
            self.assertIn("--repair", diagnosed.stdout)
            self.assertEqual(stale.read_text(), "stale managed file\n")
            self.assertFalse((target / ".coord/VERSION").exists())
            self.assertEqual(unrelated.read_text(), "operator content\n")

            repaired = self.run_install(target, "--repair")
            self.assertEqual(repaired.returncode, 0, repaired.stderr)
            self.assertIn(f"repaired {RELEASE}", repaired.stdout)
            self.assertEqual(unrelated.read_text(), "operator content\n")

            healthy = self.run_install(target, "--doctor")
            self.assertEqual(healthy.returncode, 0, healthy.stderr)
            self.assertIn(f"installed version {RELEASE}", healthy.stdout)
            self.assertIn(f"installation matches {RELEASE}", healthy.stdout)

            version = subprocess.run(
                [str(target / ".coord/coord"), "--version"],
                cwd=target,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(version.returncode, 0, version.stderr)
            self.assertEqual(version.stdout.strip(), f"coord {RELEASE}")
            installer_version = subprocess.run(
                [str(target / ".coord/install"), "--version"],
                cwd=target,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(installer_version.returncode, 0, installer_version.stderr)
            self.assertEqual(
                installer_version.stdout.strip(), f"coord-install {RELEASE}"
            )

    def test_legacy_upgrade_preserves_hooks_identity_and_both_entrypoints(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            target = Path(temporary)
            legacy = target / ".claude/coord"
            legacy.mkdir(parents=True)
            (legacy / "project-id").write_text("legacy-project\n")
            (target / "AGENTS.md").write_text("target agents\n")
            (target / "CLAUDE.md").write_text("target claude\n")
            unrelated_skill = target / ".claude/skills/local/SKILL.md"
            unrelated_skill.parent.mkdir(parents=True)
            unrelated_skill.write_text("target skill\n")
            (target / ".claude/settings.json").write_text(
                json.dumps(
                    {
                        "effortLevel": "high",
                        "hooks": {
                            "PreToolUse": [
                                {
                                    "matcher": "Bash",
                                    "hooks": [
                                        {
                                            "type": "command",
                                            "command": "printf unrelated",
                                            "timeout": 2,
                                        },
                                        {
                                            "type": "command",
                                            "command": "python3 .coord/custom-security.py",
                                            "timeout": 3,
                                        },
                                        {
                                            "type": "command",
                                            "command": "python3 .claude/coord/hook.py",
                                            "timeout": 10,
                                        },
                                    ],
                                }
                            ]
                        },
                    }
                )
            )
            (target / ".codex").mkdir()
            (target / ".codex/hooks.json").write_text(
                json.dumps(
                    {
                        "hooks": {
                            "SessionStart": [
                                {
                                    "hooks": [
                                        {
                                            "type": "command",
                                            "command": "python3 .claude/coord/session.py",
                                            "timeout": 10,
                                        }
                                    ]
                                }
                            ]
                        }
                    }
                )
            )

            installed = self.run_install(target)
            self.assertEqual(installed.returncode, 0, installed.stderr)
            self.assertEqual(
                (target / ".coord/project-id").read_text().strip(),
                "legacy-project",
            )
            self.assertEqual(
                (target / ".claude/coord/project-id").read_text().strip(),
                "legacy-project",
            )
            self.assertEqual((target / "AGENTS.md").read_text(), "target agents\n")
            self.assertEqual((target / "CLAUDE.md").read_text(), "target claude\n")
            self.assertEqual(unrelated_skill.read_text(), "target skill\n")
            settings = json.loads((target / ".claude/settings.json").read_text())
            self.assertEqual(settings["effortLevel"], "high")
            commands = [
                hook["command"]
                for group in settings["hooks"]["PreToolUse"]
                for hook in group["hooks"]
            ]
            self.assertIn("printf unrelated", commands)
            self.assertIn("python3 .coord/custom-security.py", commands)
            self.assertEqual(
                sum("/.coord/hook.py" in command for command in commands), 1
            )
            self.assertFalse(any(".claude/coord/" in command for command in commands))
            self.assertEqual(
                (target / ".claude/coord/.gitignore").read_text(),
                "state/\n__pycache__/\n",
            )
            self.assertTrue(
                (target / ".claude/coord/DESIGN.md")
                .read_text()
                .endswith("DESIGN.md).\n")
            )

            canonical = subprocess.run(
                [str(target / ".coord/coord"), "--help"],
                cwd="/",
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            compatibility = subprocess.run(
                [str(target / ".claude/coord/coord"), "--help"],
                cwd="/",
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(canonical.returncode, 0, canonical.stderr)
            self.assertEqual(compatibility.returncode, canonical.returncode)
            self.assertEqual(compatibility.stdout, canonical.stdout)

            store_path = target / ".claude/coord/store.py"
            spec = importlib.util.spec_from_file_location("legacy_store_test", store_path)
            assert spec is not None and spec.loader is not None
            module = importlib.util.module_from_spec(spec)
            sys.modules[spec.name] = module
            spec.loader.exec_module(module)
            self.assertEqual(
                module.friendly_handle("identity", "runtime").count("-"), 4
            )

            conflicting = self.run_install(
                target, "--project-id", "different-project"
            )
            self.assertEqual(conflicting.returncode, 2)
            self.assertIn("refusing project ID change", conflicting.stderr)

            (target / ".coord/coord").write_text("# drift\n")
            drift = self.run_install(target, "--check")
            self.assertEqual(drift.returncode, 1)
            self.assertIn("drift", drift.stdout)
            repaired = self.run_install(target)
            self.assertEqual(repaired.returncode, 0, repaired.stderr)
            final = self.run_install(target, "--check")
            self.assertEqual(final.returncode, 0, final.stdout + final.stderr)

    def test_installed_cli_smokes_hello_claim_and_work_create(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            target = root / "project"
            subprocess.run(
                ["jj", "git", "init", "--colocate", str(target)],
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True,
            )
            subprocess.run(
                ["jj", "-R", str(target), "bookmark", "create", "master", "-r", "@"],
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True,
            )
            installed = self.run_install(
                target, "--project-id", "smoke-project"
            )
            self.assertEqual(installed.returncode, 0, installed.stderr)
            cli = target / ".coord/coord"
            env = {
                **os.environ,
                "COORD_DB": str(root / "state/coord.sqlite3"),
                "COORD_PROJECT_ID": "smoke-project",
            }
            hello = subprocess.run(
                [
                    str(cli),
                    "hello",
                    "--conversation",
                    "smoke-conversation",
                    "--harness",
                    "test",
                    "--cwd",
                    str(target),
                ],
                cwd=target,
                env=env,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(hello.returncode, 0, hello.stderr)
            runtime = hello.stdout.strip()
            claimed = subprocess.run(
                [str(cli), "-i", runtime, "claim", "smoke/resource"],
                cwd="/",
                env=env,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(claimed.returncode, 0, claimed.stderr)
            created = subprocess.run(
                [
                    str(target / ".claude/coord/coord"),
                    "-i",
                    runtime,
                    "work",
                    "create",
                    "smoke-work",
                    "--source",
                    str(target),
                    "--scratch-root",
                    str(root / "scratch"),
                    "--json",
                ],
                cwd="/",
                env=env,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            self.assertEqual(created.returncode, 0, created.stderr)
            record = json.loads(created.stdout)
            self.assertEqual(record["task"], "smoke-work")
            self.assertTrue(Path(record["actual_path"]).is_dir())
            self.assertEqual(
                record["jj_workspace_name"], f"agent-{record['instance_id']}"
            )
            self.assertEqual(
                record["jj_bookmark"],
                f"agent/smoke-work/{record['instance_id']}",
            )
            workspace_list = subprocess.run(
                ["jj", "-R", record["actual_path"], "workspace", "list"],
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True,
            )
            self.assertIn(record["jj_workspace_name"], workspace_list.stdout)
            described = subprocess.run(
                [
                    "jj",
                    "-R",
                    record["actual_path"],
                    "log",
                    "--no-graph",
                    "-r",
                    record["jj_bookmark"],
                    "-T",
                    'description.first_line() ++ "\\n"',
                ],
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True,
            )
            self.assertEqual(
                described.stdout.strip(), record["jj_change_description"]
            )
            (Path(record["actual_path"]) / "agent-change.txt").write_text(
                "frozen candidate\n", encoding="utf-8"
            )
            for action in ("enqueue", "acquire", "release"):
                queued = subprocess.run(
                    [
                        str(cli),
                        "-i",
                        runtime,
                        "work",
                        "queue",
                        record["instance_id"],
                        action,
                    ],
                    cwd=target,
                    env=env,
                    text=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                )
                self.assertEqual(queued.returncode, 0, queued.stderr)
            inspected = subprocess.run(
                [
                    str(cli),
                    "-i",
                    runtime,
                    "work",
                    "inspect",
                    record["instance_id"],
                    "--json",
                ],
                cwd=target,
                env=env,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True,
            )
            acquired = json.loads(inspected.stdout)
            self.assertEqual(
                acquired["integration_candidate"]["commit_id"],
                acquired["change"]["commit_id"],
            )
            self.assertEqual(
                acquired["integration_lock_release_evidence"]["resource"],
                "integration/master",
            )

    def test_invalid_hook_json_refuses_before_any_target_replacement(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            target = Path(temporary)
            legacy = target / ".claude/coord"
            legacy.mkdir(parents=True)
            (legacy / "project-id").write_text("preflight-project\n")
            sentinel = legacy / "coord"
            sentinel.write_text("legacy-sentinel\n")
            (target / ".claude/settings.json").write_text("{ invalid")
            unrelated_skill = target / ".agents/skills/local/SKILL.md"
            unrelated_skill.parent.mkdir(parents=True)
            unrelated_skill.write_text("local-sentinel\n")

            refused = self.run_install(target)
            self.assertEqual(refused.returncode, 2)
            self.assertIn("invalid JSON", refused.stderr)
            self.assertEqual(sentinel.read_text(), "legacy-sentinel\n")
            self.assertFalse((target / ".coord").exists())
            self.assertFalse((target / ".codex").exists())
            self.assertEqual(unrelated_skill.read_text(), "local-sentinel\n")


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