371 lines
16 KiB
Python
Executable file
371 lines
16 KiB
Python
Executable file
#!/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())
|
|
|
|
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)
|