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