claire/tests/test_orchestrator_bootstrap.py

"""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