Merge branch 'feat/odysseus-dev-worktree-boot' into lane/decomposition-and-test-isolation

tests/cli/test_dev_cli_isolation.py

@@ -0,0 +1,171 @@
+"""The isolation contract of `odysseus dev`.
+
+The launcher exists so that two checkouts on one machine cannot share
+runtime state by accident. Every test here pins one of the guarantees
+that makes that true: derived ports never land on a port the project
+already means something by, a ChromaDB we did not start is refused
+rather than adopted, a checkout wired into a service manager is not
+bootable, and nothing is signalled on the strength of a pid alone.
+"""
+import argparse
+import os
+import socket
+
+import pytest
+
+from tests.helpers.cli_loader import load_script
+
+
+@pytest.fixture
+def cli():
+    return load_script("odysseus-dev")
+
+
+@pytest.fixture
+def worktree(tmp_path):
+    """A directory shaped enough like a checkout for the launcher to accept it."""
+    for marker in ("app.py", "setup.py", "requirements.txt"):
+        (tmp_path / marker).write_text("")
+    (tmp_path / "venv" / "bin").mkdir(parents=True)
+    (tmp_path / "venv" / "bin" / "python").write_text("")
+    return tmp_path
+
+
+def up_args(**overrides):
+    defaults = dict(
+        port=None, chroma_port=None, no_chroma=False, venv=None, from_pr=None,
+        remote="origin", foreground=False, timeout=5, pretty=False,
+    )
+    defaults.update(overrides)
+    return argparse.Namespace(**defaults)
+
+
+def test_derived_ports_are_stable_distinct_and_never_reserved(cli):
+    first = cli.derive_ports("/checkouts/alpha")
+    assert first == cli.derive_ports("/checkouts/alpha")
+    assert first != cli.derive_ports("/checkouts/beta")
+    assert first["chroma"] == first["app"] + 1
+    assert first["test_static"] == first["app"] + 2
+
+    # No path can derive onto a port the project already owns — 7860 is a
+    # normal start-macos.sh launch, 8100 somebody else's vector store.
+    for index in range(500):
+        for port in cli.derive_ports(f"/checkouts/w{index}").values():
+            assert cli.reserved_reason(port) is None, port
+
+
+def test_reserved_ports_are_refused_even_when_asked_for(cli, worktree, monkeypatch):
+    monkeypatch.chdir(worktree)
+    with pytest.raises(SystemExit):
+        cli.resolve_ports(worktree, up_args(port=7860))
+    with pytest.raises(SystemExit):
+        cli.resolve_ports(worktree, up_args(chroma_port=8100))
+
+
+def test_root_is_resolved_from_the_working_directory(cli, worktree):
+    nested = worktree / "static" / "js"
+    nested.mkdir(parents=True)
+    assert cli.find_repo_root(nested) == worktree.resolve()
+    assert cli.find_repo_root(worktree.parent) is None
+
+
+def test_a_checkout_run_by_a_service_manager_is_not_bootable(cli, worktree, monkeypatch):
+    units = worktree.parent / "units"
+    units.mkdir()
+    (units / "com.odysseus.server.plist").write_text(
+        f"<plist><string>{worktree.resolve()}/start-macos.sh</string></plist>"
+    )
+    assert cli.managed_by_service(worktree, unit_dirs=[units]).endswith(".plist")
+    unrelated = worktree.parent / "somewhere-else"
+    unrelated.mkdir()
+    assert cli.managed_by_service(unrelated, unit_dirs=[units]) is None
+
+    monkeypatch.chdir(worktree)
+    monkeypatch.setattr(cli, "service_unit_dirs", lambda: [units])
+    with pytest.raises(SystemExit):
+        cli.cmd_up(up_args())
+
+
+def test_a_chromadb_we_did_not_start_is_refused_not_adopted(cli, worktree, monkeypatch):
+    monkeypatch.chdir(worktree)
+    monkeypatch.setattr(cli, "service_unit_dirs", list)
+    ports = cli.derive_ports(worktree)
+
+    foreign = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    foreign.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    try:
+        foreign.bind(("127.0.0.1", ports["chroma"]))
+    except OSError:
+        pytest.skip(f"derived chroma port {ports['chroma']} is unavailable on this host")
+    foreign.listen(1)
+    try:
+        with pytest.raises(SystemExit):
+            cli.cmd_up(up_args())
+    finally:
+        foreign.close()
+
+    # Refused means refused: nothing was started and no state was recorded.
+    assert not cli.state_path(worktree).exists()
+
+
+def test_no_chroma_points_the_app_at_a_port_nothing_answers(cli):
+    port = cli.unused_port()
+    assert not cli.port_bound(port)
+    assert cli.reserved_reason(port) is None
+
+
+def test_every_chroma_outcome_leaves_the_app_off_a_port_we_do_not_own(cli, worktree):
+    """The decision has three endings and none of them is "use theirs"."""
+    ports = cli.derive_ports(worktree)
+    data, logs = worktree / "data", worktree / "logs"
+    data.mkdir()
+    logs.mkdir()
+    venv_python = worktree / "venv" / "bin" / "python"
+
+    # 1. Asked to go without: a port nothing answers on, not the derived
+    # one, which is where a foreign server may appear later.
+    entry, port, note = cli.resolve_chroma(
+        up_args(no_chroma=True), {}, ports, venv_python, data, logs
+    )
+    assert (entry, port != ports["chroma"], cli.port_bound(port)) == (None, True, False)
+    assert "no-chroma" in note
+
+    # 2. No server to start (this venv has no `chroma` binary, which is
+    # the stock requirements.txt): keyword mode, and again not the
+    # derived port.
+    entry, port, note = cli.resolve_chroma(
+        up_args(), {}, ports, venv_python, data, logs
+    )
+    assert entry is None and port != ports["chroma"]
+    assert "keyword-only" in note
+
+
+def test_a_pid_is_never_trusted_without_its_command_line(cli, worktree, monkeypatch):
+    own_pid = os.getpid()
+    assert cli.pid_is_ours(own_pid, ["definitely-not-in-this-command-line"]) is False
+    assert cli.pid_is_ours(None, []) is False
+    assert cli.pid_is_ours(own_pid, [cli.pid_command(own_pid).split()[0]]) is True
+
+    # `down` must not signal a live process whose fingerprints disagree —
+    # here, this very test run — and must keep the record so the pid can
+    # be investigated rather than lost.
+    cli.write_state(worktree, {"app": {"pid": own_pid, "fingerprints": ["uvicorn --port 1"]}})
+    monkeypatch.chdir(worktree)
+    monkeypatch.setattr(os, "kill", _forbidden_kill)
+    cli.cmd_down(up_args())
+    assert cli.state_path(worktree).exists()
+
+
+def test_down_forgets_an_instance_that_is_gone(cli, worktree, monkeypatch):
+    cli.write_state(worktree, {"app": {"pid": 2 ** 31 - 1, "fingerprints": ["uvicorn"]}})
+    monkeypatch.chdir(worktree)
+    cli.cmd_down(up_args())
+    assert not cli.state_path(worktree).exists()
+
+
+def _forbidden_kill(pid, sig):
+    """Liveness probes (signal 0) are fine; anything that would actually
+    reach the process is the failure this test is about."""
+    if sig == 0:
+        return None
+    raise AssertionError(f"cmd_down sent signal {sig} to a process it does not own")