225 lines
7 KiB
Python
225 lines
7 KiB
Python
"""Unit tests for the orchestrator bootstrap module + CLI command."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid as _uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from claire.cli import app as claire_app
|
|
from claire.orchestrator import bootstrap
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_cfg(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""Pin claire.toml to a tmp_path."""
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data"))
|
|
return tmp_path / "config" / "claire" / "claire.toml"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _client_host
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("bind", "expected"),
|
|
[
|
|
("0.0.0.0", "127.0.0.1"),
|
|
("::", "127.0.0.1"),
|
|
("", "127.0.0.1"),
|
|
("127.0.0.1", "127.0.0.1"),
|
|
("10.9.0.2", "10.9.0.2"),
|
|
("apricot", "apricot"),
|
|
],
|
|
)
|
|
def test_client_host_collapses_wildcard_binds(bind: str, expected: str) -> None:
|
|
assert bootstrap._client_host(bind) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_workspace
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_write_workspace_creates_mcp_json_and_claude_md(tmp_path: Path) -> None:
|
|
cwd = tmp_path / "orch"
|
|
bootstrap.write_workspace(cwd, "http://127.0.0.1:8765/mcp/sse")
|
|
assert (cwd / ".mcp.json").exists()
|
|
assert (cwd / "CLAUDE.md").exists()
|
|
assert "127.0.0.1:8765" in (cwd / ".mcp.json").read_text()
|
|
assert "submit_chat_reply" in (cwd / "CLAUDE.md").read_text()
|
|
|
|
|
|
def test_write_workspace_is_idempotent(tmp_path: Path) -> None:
|
|
cwd = tmp_path / "orch"
|
|
bootstrap.write_workspace(cwd, "url-a")
|
|
bootstrap.write_workspace(cwd, "url-b") # rewrites with new url
|
|
assert "url-b" in (cwd / ".mcp.json").read_text()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_session_uuid / clear_session_uuid
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_write_session_uuid_persists_to_config(isolated_cfg: Path) -> None:
|
|
sid = str(_uuid.uuid4())
|
|
cfg = bootstrap.write_session_uuid(sid, host="local")
|
|
assert cfg.orchestrator.session_uuid == sid
|
|
# Round-trip via load_or_init.
|
|
from claire.config import load_or_init
|
|
reloaded = load_or_init()
|
|
assert reloaded.orchestrator.session_uuid == sid
|
|
|
|
|
|
def test_write_session_uuid_rejects_non_uuid(isolated_cfg: Path) -> None:
|
|
with pytest.raises(bootstrap.BootstrapError):
|
|
bootstrap.write_session_uuid("not-a-uuid")
|
|
|
|
|
|
def test_clear_session_uuid_resets_to_default(isolated_cfg: Path) -> None:
|
|
bootstrap.write_session_uuid(str(_uuid.uuid4()))
|
|
cfg = bootstrap.clear_session_uuid()
|
|
assert cfg.orchestrator.session_uuid is None
|
|
assert cfg.orchestrator.host == "local"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_session — fake rclaude
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class _FakeRow:
|
|
host: str
|
|
uuid: UUID
|
|
cwd: str
|
|
mtime_epoch: int
|
|
snippet: str = ""
|
|
|
|
|
|
class _FakeRclaude:
|
|
def __init__(self, rows: list[_FakeRow]) -> None:
|
|
self._rows = rows
|
|
self.calls = 0
|
|
|
|
def list_sessions(self) -> list[_FakeRow]:
|
|
self.calls += 1
|
|
return self._rows
|
|
|
|
|
|
def test_discover_session_picks_match_in_cwd(tmp_path: Path) -> None:
|
|
cwd = tmp_path / "orch"
|
|
cwd.mkdir()
|
|
matching = _uuid.uuid4()
|
|
other = _uuid.uuid4()
|
|
rcl = _FakeRclaude([
|
|
_FakeRow(host="local", uuid=other, cwd="/somewhere/else", mtime_epoch=100),
|
|
_FakeRow(host="local", uuid=matching, cwd=str(cwd), mtime_epoch=200),
|
|
])
|
|
out = bootstrap.discover_session(
|
|
cwd=cwd, host="local", rclaude=rcl, timeout_s=1, poll_interval_s=0.05,
|
|
)
|
|
assert out == str(matching)
|
|
|
|
|
|
def test_discover_session_returns_none_on_timeout(tmp_path: Path) -> None:
|
|
cwd = tmp_path / "orch"
|
|
cwd.mkdir()
|
|
rcl = _FakeRclaude([])
|
|
out = bootstrap.discover_session(
|
|
cwd=cwd, host="local", rclaude=rcl, timeout_s=0.3, poll_interval_s=0.1,
|
|
)
|
|
assert out is None
|
|
|
|
|
|
def test_discover_session_prefers_most_recent_on_tie(tmp_path: Path) -> None:
|
|
cwd = tmp_path / "orch"
|
|
cwd.mkdir()
|
|
older = _uuid.uuid4()
|
|
newer = _uuid.uuid4()
|
|
rcl = _FakeRclaude([
|
|
_FakeRow(host="local", uuid=older, cwd=str(cwd), mtime_epoch=100),
|
|
_FakeRow(host="local", uuid=newer, cwd=str(cwd), mtime_epoch=999),
|
|
])
|
|
out = bootstrap.discover_session(
|
|
cwd=cwd, rclaude=rcl, timeout_s=1, poll_interval_s=0.05,
|
|
)
|
|
assert out == str(newer)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# init() integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_init_with_explicit_uuid_skips_discovery(
|
|
isolated_cfg: Path, tmp_path: Path,
|
|
) -> None:
|
|
sid = str(_uuid.uuid4())
|
|
cwd = tmp_path / "orch"
|
|
result = bootstrap.init(
|
|
cwd=cwd, session_uuid=sid, auto_discover=False,
|
|
)
|
|
assert result.session_uuid == sid
|
|
assert (cwd / ".mcp.json").exists()
|
|
# Config persisted.
|
|
from claire.config import load_or_init
|
|
assert load_or_init().orchestrator.session_uuid == sid
|
|
|
|
|
|
def test_init_without_uuid_no_discover_prints_instructions(
|
|
isolated_cfg: Path, tmp_path: Path,
|
|
) -> None:
|
|
cwd = tmp_path / "orch"
|
|
result = bootstrap.init(cwd=cwd, auto_discover=False)
|
|
assert result.session_uuid is None
|
|
assert "tmux new-session" in result.instructions
|
|
assert "Workspace scaffold ready" in result.instructions
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cli_orchestrator_show_unconfigured(isolated_cfg: Path) -> None:
|
|
runner = CliRunner()
|
|
r = runner.invoke(claire_app, ["orchestrator", "show"])
|
|
assert r.exit_code == 0
|
|
assert "not configured" in r.stdout
|
|
|
|
|
|
def test_cli_orchestrator_init_with_uuid_persists(
|
|
isolated_cfg: Path, tmp_path: Path,
|
|
) -> None:
|
|
runner = CliRunner()
|
|
sid = str(_uuid.uuid4())
|
|
r = runner.invoke(claire_app, [
|
|
"orchestrator", "init",
|
|
"--session-uuid", sid,
|
|
"--cwd", str(tmp_path / "orch"),
|
|
])
|
|
assert r.exit_code == 0, r.stdout
|
|
# Verify show now reports it.
|
|
r2 = runner.invoke(claire_app, ["orchestrator", "show"])
|
|
assert sid in r2.stdout
|
|
|
|
|
|
def test_cli_orchestrator_reset_clears(isolated_cfg: Path) -> None:
|
|
runner = CliRunner()
|
|
runner.invoke(claire_app, [
|
|
"orchestrator", "init",
|
|
"--session-uuid", str(_uuid.uuid4()),
|
|
"--no-discover",
|
|
])
|
|
r = runner.invoke(claire_app, ["orchestrator", "reset"])
|
|
assert r.exit_code == 0
|
|
r2 = runner.invoke(claire_app, ["orchestrator", "show"])
|
|
assert "not configured" in r2.stdout
|