diff --git a/conftest.py b/conftest.py index 88a2656d29..c8190aeb0e 100644 --- a/conftest.py +++ b/conftest.py @@ -11,12 +11,17 @@ from __future__ import annotations import functools +import json +import os +import pathlib import shutil import typing as t +import uuid import pytest from _pytest.doctest import DoctestItem +from libtmux._arena import ARENA_EXCLUDED_SOURCES, ArenaSpec from libtmux._internal.control_mode import ControlMode from libtmux.client import Client from libtmux.pane import Pane @@ -25,48 +30,349 @@ from libtmux.session import Session from libtmux.window import Window -if t.TYPE_CHECKING: - import pathlib - pytest_plugins = ["pytester"] +ARENA_EVIDENCE_PREFIX = "LIBTMUX_ARENA_EVIDENCE=" +# One round-trip answers identity and the challenge together: pid and +# socket_path prove *which* server answered, @libtmux_arena_challenge proves +# it is still configured as the one that was lent (a silently spawned +# replacement starts with no such option set). +ARENA_IDENTITY_FORMAT = "#{pid}\t#{socket_path}\t#{@libtmux_arena_challenge}" +ARENA_SPEC_KEY: pytest.StashKey[ArenaSpec] = pytest.StashKey() +# A set rather than one path: an artifact may audit several pages, and the +# supervisor then expects one evidence record per page. +ARENA_TARGETS_KEY: pytest.StashKey[frozenset[pathlib.Path]] = pytest.StashKey() +ARENA_DISCOVERED_PATHS_KEY: pytest.StashKey[frozenset[pathlib.Path]] = pytest.StashKey() +# Node ids per source rather than one flat set, so a record can name the page +# that produced it, and a page that collected nothing is caught. +ARENA_DISCOVERED_KEY: pytest.StashKey[dict[pathlib.Path, frozenset[str]]] = ( + pytest.StashKey() +) +ARENA_COLLECTED_KEY: pytest.StashKey[dict[pathlib.Path, frozenset[str]]] = ( + pytest.StashKey() +) +ARENA_PASSED_KEY: pytest.StashKey[dict[pathlib.Path, frozenset[str]]] = ( + pytest.StashKey() +) +# The identity captured once, before any page runs, so every later query has +# something fixed to compare against rather than just "looks fine to itself". +ARENA_IDENTITY_KEY: pytest.StashKey[tuple[int, str, str]] = pytest.StashKey() +# Session names already on the lend before this run touched it, so teardown +# can reap exactly what the run's own examples left behind and nothing else. +ARENA_BASELINE_SESSIONS_KEY: pytest.StashKey[frozenset[str]] = pytest.StashKey() +# The last node id collected per source, in the order pytest will actually +# run them -- how the per-page identity check knows a page has finished +# without guessing at item counts. +ARENA_LAST_NODEID_KEY: pytest.StashKey[dict[pathlib.Path, str]] = pytest.StashKey() + + +def _arena_spec(config: pytest.Config) -> ArenaSpec | None: + """Return the validated arena contract for this pytest invocation.""" + return config.stash.get(ARENA_SPEC_KEY, None) + + +def _arena_targets(config: pytest.Config) -> frozenset[pathlib.Path] | None: + """Return the validated arena sources for this pytest invocation.""" + return config.stash.get(ARENA_TARGETS_KEY, None) + + +def _query_arena_identity(spec: ArenaSpec) -> tuple[int, str, str]: + """Return (pid, socket_path, challenge) for the endpoint ``spec`` names. + + ``display-message`` only ever answers a server already listening on the + requested socket -- unlike ``new-session``, tmux does not spawn one to + service it (verified: a `list-sessions`/`display-message` against a + socket with nothing listening just errors, it never creates the socket + file). So this query itself can never be what silently starts the + replacement server a stopped lend produces. + """ + server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) + result = server.cmd("display-message", "-p", ARENA_IDENTITY_FORMAT).stdout + if len(result) != 1: + msg = "arena server identity query returned an unexpected result" + raise RuntimeError(msg) + parts = result[0].split("\t", 2) + if len(parts) != 3 or parts[1] != spec.socket_path or not parts[2]: + msg = "arena server identity does not match the requested endpoint" + raise RuntimeError(msg) + return int(parts[0]), parts[1], parts[2] + + +def _verify_arena_identity( + spec: ArenaSpec, + baseline: tuple[int, str, str], + where: str, +) -> None: + """Raise, naming ``where``, if the lent server's identity has moved on. + + ``where`` is the source whose examples just finished running, or "the + run" for the check just before evidence is published -- so a break is + attributed to the page after which it was detected, rather than only + discovered once every requested page has already run against whatever + answered next. + """ + try: + observed = _query_arena_identity(spec) + except RuntimeError as exc_info: + msg = f"arena server identity check failed after {where!r}: {exc_info}" + raise RuntimeError(msg) from exc_info + if observed != baseline: + msg = ( + f"arena server identity changed after {where!r}: expected " + f"pid/socket/challenge {baseline!r}, got {observed!r}" + ) + raise RuntimeError(msg) + + +def _reap_arena_sessions(spec: ArenaSpec, baseline_sessions: frozenset[str]) -> None: + """Kill every session this run's own examples left behind. + + Idempotent: a session already gone (already reaped, or never existed) + is not an error. Scoped to the diff against ``baseline_sessions`` -- the + lend's own sessions from before this run touched it -- so a session that + predates the run, such as its hold session, is never a candidate. + """ + server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) + current = {s.session_name for s in server.sessions if s.session_name is not None} + for name in sorted(current - baseline_sessions): + cleanup = server.cmd("kill-session", target=name) + already_gone = cleanup.returncode != 0 and any( + "can't find session" in line or "session not found" in line + for line in cleanup.stderr + ) + if cleanup.returncode != 0 and not already_gone: + msg = f"arena teardown could not reap leaked session {name!r}" + raise RuntimeError(msg) + + +def pytest_addoption(parser: pytest.Parser) -> None: + """Register the source(s) selected by the arena adapter. + + Repeatable: one flag per audited page. The adapter passes exactly the + pages its artifact declares, and nothing else may be collected. + """ + parser.addoption( + "--libtmux-arena-target", + metavar="PATH", + action="append", + help="Run one audited doctest source against an external tmux server " + "(repeatable)", + ) + + +def pytest_configure(config: pytest.Config) -> None: + """Validate the arena contract before pytest initializes fixtures.""" + try: + spec = ArenaSpec.from_environ(os.environ) + except ValueError as exc_info: + raise pytest.UsageError(str(exc_info)) from exc_info + if spec is None: + return + + raw_targets = frozenset(config.getoption("libtmux_arena_target") or []) + excluded = { + source: ARENA_EXCLUDED_SOURCES[source] + for source in raw_targets + if source in ARENA_EXCLUDED_SOURCES + } + if excluded: + # Checked before the artifact's own tuple is even consulted: a + # source like this must be refused by name, not merely absent from + # ARENA_ARTIFACT_TARGETS -- the two look identical from outside a + # mismatch error otherwise. + reasons = "; ".join( + f"{source!r} ({reason})" for source, reason in sorted(excluded.items()) + ) + msg = f"arena refuses excluded source(s): {reasons}" + raise pytest.UsageError(msg) + + expected_relative = frozenset( + p.as_posix() for p in spec.targets_for(pathlib.Path()) + ) + if raw_targets != expected_relative: + msg = f"arena artifact {spec.artifact!r} requires {sorted(expected_relative)!r}" + raise pytest.UsageError(msg) + root = pathlib.Path(config.rootpath).resolve() + targets = frozenset(p.resolve(strict=True) for p in spec.targets_for(root)) + config.stash[ARENA_SPEC_KEY] = spec + config.stash[ARENA_TARGETS_KEY] = targets + + # Baseline, captured before any page runs: what "the lent server" and + # "its sessions" mean for the rest of this run. Every later identity + # check compares against this rather than against its own last query, so + # a slow drift across several pages cannot pass by always comparing + # favorably to the most recent (possibly already-wrong) reading. + try: + config.stash[ARENA_IDENTITY_KEY] = _query_arena_identity(spec) + except RuntimeError as exc_info: + raise pytest.UsageError(str(exc_info)) from exc_info + baseline_server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) + config.stash[ARENA_BASELINE_SESSIONS_KEY] = frozenset( + s.session_name for s in baseline_server.sessions if s.session_name is not None + ) + + +def pytest_collection_finish(session: pytest.Session) -> None: + """Reject selections that include anything besides the audited sources.""" + targets = _arena_targets(session.config) + if targets is None: + return + paths = {item.path.resolve() for item in session.items} + selected = frozenset(item.nodeid for item in session.items) + discovered_paths = session.config.stash.get(ARENA_DISCOVERED_PATHS_KEY, frozenset()) + discovered = session.config.stash.get(ARENA_DISCOVERED_KEY, {}) + discovered_all = ( + frozenset().union(*discovered.values()) if discovered else frozenset() + ) + if ( + paths != targets + or discovered_paths != targets + or discovered.keys() != targets + or any(not nodeids for nodeids in discovered.values()) + or selected != discovered_all + ): + msg = "arena requires collection of exactly the audited doctest sources" + raise pytest.UsageError(msg) + session.config.stash[ARENA_COLLECTED_KEY] = dict(discovered) + + # The last node id per source, in the order pytest is actually about to + # run them (not the order they happened to be discovered in). This is + # how the per-page identity check recognizes "a page just finished" + # without hard-coding how many items any given page collects. + last_nodeid_by_path: dict[pathlib.Path, str] = {} + for item in session.items: + item_path = item.path.resolve() + if item_path in targets: + last_nodeid_by_path[item_path] = item.nodeid + session.config.stash[ARENA_LAST_NODEID_KEY] = last_nodeid_by_path + + +def pytest_itemcollected(item: pytest.Item) -> None: + """Record every arena item before pytest applies filters.""" + targets = _arena_targets(item.config) + if targets is None: + return + path = item.path.resolve() + discovered_paths = item.config.stash.get(ARENA_DISCOVERED_PATHS_KEY, frozenset()) + item.config.stash[ARENA_DISCOVERED_PATHS_KEY] = discovered_paths | {path} + if path in targets: + discovered = item.config.stash.get(ARENA_DISCOVERED_KEY, {}) + discovered = dict(discovered) + discovered[path] = discovered.get(path, frozenset()) | {item.nodeid} + item.config.stash[ARENA_DISCOVERED_KEY] = discovered + + +def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[t.Any]) -> None: + """Record successful arena doctest calls for evidence publication.""" + targets = _arena_targets(item.config) + path = item.path.resolve() + if targets is None or path not in targets: + return + if call.when == "call" and call.excinfo is None: + passed = item.config.stash.get(ARENA_PASSED_KEY, {}) + passed = dict(passed) + passed[path] = passed.get(path, frozenset()) | {item.nodeid} + item.config.stash[ARENA_PASSED_KEY] = passed + @pytest.fixture(autouse=True) def add_doctest_fixtures( request: pytest.FixtureRequest, doctest_namespace: dict[str, t.Any], -) -> None: +) -> t.Generator[None]: """Configure doctest fixtures for pytest-doctest.""" - if isinstance(request._pyfuncitem, DoctestItem) and shutil.which("tmux"): + if not isinstance(request._pyfuncitem, DoctestItem): + yield + return + + spec = _arena_spec(request.config) + if spec is None and not shutil.which("tmux"): + yield + return + + if spec is None: request.getfixturevalue("set_home") - doctest_namespace["Server"] = Server - doctest_namespace["Session"] = Session - doctest_namespace["Window"] = Window - doctest_namespace["Pane"] = Pane - doctest_namespace["Client"] = Client - doctest_namespace["server"] = request.getfixturevalue("server") - doctest_namespace["Server"] = request.getfixturevalue("TestServer") + server = request.getfixturevalue("server") + test_server = request.getfixturevalue("TestServer") session: Session = request.getfixturevalue("session") - doctest_namespace["session"] = session - doctest_namespace["window"] = session.active_window - doctest_namespace["pane"] = session.active_pane - doctest_namespace["request"] = request - doctest_namespace["ControlMode"] = ControlMode - doctest_namespace["control_mode"] = functools.partial( - ControlMode, - server=session.server, - session=session, + else: + server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) + session_name = f"libtmux_arena_{uuid.uuid4().hex}" + session = server.new_session(session_name=session_name) + test_server = functools.partial( + Server, + socket_path=spec.socket_path, + tmux_bin=spec.tmux_bin, ) - doctest_namespace["monkeypatch"] = request.getfixturevalue("monkeypatch") + + doctest_namespace["Server"] = Server + doctest_namespace["Session"] = Session + doctest_namespace["Window"] = Window + doctest_namespace["Pane"] = Pane + doctest_namespace["Client"] = Client + doctest_namespace["server"] = server + doctest_namespace["Server"] = test_server + doctest_namespace["session"] = session + doctest_namespace["window"] = session.active_window + doctest_namespace["pane"] = session.active_pane + doctest_namespace["request"] = request + doctest_namespace["ControlMode"] = ControlMode + doctest_namespace["control_mode"] = functools.partial( + ControlMode, + server=session.server, + session=session, + ) + doctest_namespace["monkeypatch"] = request.getfixturevalue("monkeypatch") + try: + yield + finally: + if spec is not None: + cleanup = server.cmd("kill-session", target=session_name) + # idempotent teardown. A doctest that kills its own + # last pane/window (tmux: last window dies -> session dies) has + # already achieved this fixture's goal -- "our session is gone" + # -- by the time we get here. Only a session that is *not* gone + # and *failed* to go is a real cleanup failure; do not conflate + # "already torn down" with "could not tear down" (see + # pane_interaction.md:467,477 in borrowed.md finding 2). + already_gone = cleanup.returncode != 0 and any( + "can't find session" in line or "session not found" in line + for line in cleanup.stderr + ) + if cleanup.returncode != 0 and not already_gone: + msg = "arena session cleanup failed" + raise RuntimeError(msg) + + # Prove identity survived this page before the next one starts + # to run against whatever answers next. Only the item that is + # last (in run order) for its source checks -- an earlier item + # in the same page would just repeat a check its own page + # hasn't finished yet, and the page boundary is exactly where a + # doctest-content break (like the excluded context_managers.md's + # `with Server()`) would land. + last_nodeid_by_path = request.config.stash.get( + ARENA_LAST_NODEID_KEY, + {}, + ) + item_path = request._pyfuncitem.path.resolve() + if last_nodeid_by_path.get(item_path) == request._pyfuncitem.nodeid: + baseline = request.config.stash.get(ARENA_IDENTITY_KEY, None) + if baseline is not None: + source = item_path.relative_to( + request.config.rootpath, + ).as_posix() + _verify_arena_identity(spec, baseline, source) @pytest.fixture(autouse=True) def set_home( monkeypatch: pytest.MonkeyPatch, user_path: pathlib.Path, + request: pytest.FixtureRequest, ) -> None: """Configure home directory for pytest tests.""" - monkeypatch.setenv("HOME", str(user_path)) + if _arena_spec(request.config) is None: + monkeypatch.setenv("HOME", str(user_path)) @pytest.fixture(autouse=True) @@ -84,3 +390,68 @@ def setup_session( """Session-level test configuration for pytest.""" if USING_ZSH: request.getfixturevalue("zshrc") + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """Reap the run's stray sessions, then publish evidence once all pass. + + Reaping runs whenever the identity captured at ``pytest_configure`` still + matches -- regardless of ``exitstatus`` -- so a run that fails for an + unrelated reason does not leave the lend dirtier than it found it. + It is skipped, not attempted against a guess, the moment identity no + longer matches: after a break (or a silent replacement) the socket may + not even be answering for the server this run was lent, and teardown + does not touch a tmux server it cannot first prove is that one. + + fail-closed is preserved for evidence because ``exitstatus`` is only + ``pytest.ExitCode.OK`` when every collected item passed; since collection + already required every target to contribute at least one item + (``pytest_collection_finish``), ``collected_by_target`` and + ``passed_by_target`` are equal for every target by construction once we + reach this point. No partial-source evidence is possible: either every + requested page gets a record, or (on any failure/error) none do. + """ + spec = _arena_spec(session.config) + targets = _arena_targets(session.config) + if spec is None or targets is None: + return + + baseline_identity = session.config.stash.get(ARENA_IDENTITY_KEY, None) + baseline_sessions = session.config.stash.get(ARENA_BASELINE_SESSIONS_KEY, None) + try: + observed_identity = _query_arena_identity(spec) + except RuntimeError: + observed_identity = None + identity_intact = ( + baseline_identity is not None and observed_identity == baseline_identity + ) + if identity_intact and baseline_sessions is not None: + _reap_arena_sessions(spec, baseline_sessions) + + collected_by_target = session.config.stash.get(ARENA_COLLECTED_KEY, {}) + passed_by_target = session.config.stash.get(ARENA_PASSED_KEY, {}) + if ( + exitstatus != pytest.ExitCode.OK + or not collected_by_target + or collected_by_target.keys() != targets + or passed_by_target != collected_by_target + or session.config.getoption("collectonly") + ): + return + + if baseline_identity is None or not identity_intact: + msg = "arena server identity does not match the requested endpoint" + raise RuntimeError(msg) + for target in sorted(targets, key=lambda p: p.as_posix()): + evidence = { + "artifact": spec.artifact, + "challenge": baseline_identity[2], + "schema": 1, + "server_pid": baseline_identity[0], + "socket_path": baseline_identity[1], + "source": target.relative_to(session.config.rootpath).as_posix(), + } + print( + "\n" + ARENA_EVIDENCE_PREFIX + json.dumps(evidence, sort_keys=True), + flush=True, + ) diff --git a/src/libtmux/_arena.py b/src/libtmux/_arena.py new file mode 100644 index 0000000000..2dc3895ec2 --- /dev/null +++ b/src/libtmux/_arena.py @@ -0,0 +1,89 @@ +"""Contract values for the opt-in arena doctest adapter.""" + +from __future__ import annotations + +import dataclasses +import pathlib +import typing as t + +ARENA_ARTIFACT_TARGETS = { + "python-exact-binary": ("docs/topics/workspace_setup.md",), + "python-workspace-setup": ("docs/topics/workspace_setup.md",), + # The first artifact to actually exercise several sources against one + # lent server. Both pages only ever touch the `server`/`session` the + # arena fixture hands them -- neither constructs its own `Server()`, so + # neither can reach the lent daemon's own kill-server (see + # ARENA_EXCLUDED_SOURCES below for the page that does). + "python-workspace-and-location": ( + "docs/topics/workspace_setup.md", + "docs/topics/self_location.md", + ), +} + +# A source that must never run under the arena: its examples stop the lent +# server rather than a private one. Named here so the refusal is a specific, +# visible reason rather than an absence from ARENA_ARTIFACT_TARGETS -- the +# two would otherwise look identical from the outside (a source rejected for +# not matching the requested artifact's tuple looks like a typo, not a +# safety rule). +ARENA_EXCLUDED_SOURCES: dict[str, str] = { + "docs/topics/context_managers.md": ( + "every example opens `with Server()`; in arena mode the doctest " + "namespace's `Server` name is bound to a factory pinned to the " + "lent socket path, so leaving the block runs Server.__exit__ -> " + "Server.kill() -> `kill-server` against the borrowed daemon " + "itself, not a private one" + ), +} + + +def _assert_no_excluded_targets( + artifact_targets: t.Mapping[str, tuple[str, ...]], +) -> None: + """Refuse at import time if any artifact ever names an excluded source. + + A mapping edit that adds an excluded page back in would otherwise only + surface the first time someone ran that artifact against a real lent + server -- by which point it may already have stopped it. + """ + conflicts = { + artifact: sorted(overlap) + for artifact, sources in artifact_targets.items() + if (overlap := frozenset(sources) & ARENA_EXCLUDED_SOURCES.keys()) + } + if conflicts: + msg = f"artifact(s) name an excluded arena source: {conflicts!r}" + raise AssertionError(msg) + + +_assert_no_excluded_targets(ARENA_ARTIFACT_TARGETS) + + +@dataclasses.dataclass(frozen=True) +class ArenaSpec: + """Describe one explicitly selected external tmux endpoint.""" + + artifact: str + socket_path: str + tmux_bin: str + + @classmethod + def from_environ(cls, environ: t.Mapping[str, str]) -> ArenaSpec | None: + """Return an active specification only for a complete descriptor contract.""" + if not environ.get("LIBTMUX_ARENA_DESCRIPTOR"): + return None + + artifact = environ.get("LIBTMUX_ARENA_ARTIFACT") + socket_path = environ.get("LIBTMUX_SOCKET_PATH") + tmux_bin = environ.get("LIBTMUX_TMUX_BIN") + if not artifact or not socket_path or not tmux_bin: + msg = "arena descriptor, artifact, socket, and tmux executable are required" + raise ValueError(msg) + if artifact not in ARENA_ARTIFACT_TARGETS: + msg = f"arena artifact {artifact!r} has no audited source mapping" + raise ValueError(msg) + return cls(artifact=artifact, socket_path=socket_path, tmux_bin=tmux_bin) + + def targets_for(self, root: pathlib.Path) -> tuple[pathlib.Path, ...]: + """Resolve every source bound to this artifact inside ``root``.""" + return tuple(root / rel for rel in ARENA_ARTIFACT_TARGETS[self.artifact]) diff --git a/tests/test_arena.py b/tests/test_arena.py new file mode 100644 index 0000000000..455ebb7eee --- /dev/null +++ b/tests/test_arena.py @@ -0,0 +1,652 @@ +"""Tests for the arena doctest adapter.""" + +from __future__ import annotations + +import dataclasses +import importlib +import json +import os +import pathlib +import shlex +import shutil +import subprocess +import sys +import typing as t + +import pytest + +from libtmux.server import Server + +ROOT = pathlib.Path(__file__).parents[1] +TARGET = "docs/topics/workspace_setup.md" +TWO_PAGE_ARTIFACT = "python-workspace-and-location" +FIRST_PAGE = "docs/topics/workspace_setup.md" +SECOND_PAGE = "docs/topics/self_location.md" +# Must match conftest.py's ARENA_IDENTITY_FORMAT exactly -- the format string +# the between-page identity check sends to `display-message`. +ARENA_IDENTITY_PROBE = "#{pid}\t#{socket_path}\t#{@libtmux_arena_challenge}" + + +@dataclasses.dataclass(frozen=True) +class ArenaEndpoint: + """Retain an externally owned tmux server and its hold session.""" + + server: Server + hold_name: str + socket_path: str + challenge: str | None + + +def _external_endpoint( + TestServer: t.Callable[..., Server], + *, + hold_name: str = "arena-hold", + challenge: str | None = "arena-challenge", +) -> ArenaEndpoint: + """Start one external tmux daemon that the adapter must not own.""" + server = TestServer() + server.new_session(session_name=hold_name) + socket_path = server.cmd("display-message", "-p", "#{socket_path}").stdout[0] + if challenge is not None: + server.cmd("set-option", "-g", "@libtmux_arena_challenge", challenge) + return ArenaEndpoint(server, hold_name, socket_path, challenge) + + +def _arena_environ(endpoint: ArenaEndpoint, tmux_bin: str) -> dict[str, str]: + """Build the complete activated environment for one external endpoint.""" + return os.environ | { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "python-exact-binary", + "LIBTMUX_SOCKET_PATH": endpoint.socket_path, + "LIBTMUX_TMUX_BIN": tmux_bin, + } + + +def _run_arena( + environ: dict[str, str], *arguments: str +) -> subprocess.CompletedProcess[str]: + """Run the native pytest entrypoint with the supplied arena selection.""" + return subprocess.run( + [sys.executable, "-m", "pytest", "--reruns=0", *arguments], + capture_output=True, + check=False, + cwd=ROOT, + env=environ, + text=True, + ) + + +def _assert_only_hold(endpoint: ArenaEndpoint) -> None: + """Assert the adapter did not own or retain the external daemon state.""" + assert endpoint.server.is_alive() + assert [session.session_name for session in endpoint.server.sessions] == [ + endpoint.hold_name + ] + + +def _remove_adapter_sessions(endpoint: ArenaEndpoint) -> None: + """Remove only sessions that a failed adapter run could have created.""" + for session in endpoint.server.sessions: + session_name = session.session_name + if session_name is not None and session_name != endpoint.hold_name: + endpoint.server.kill_session(session_name) + + +def _failing_tmux_wrapper( + tmp_path: pathlib.Path, + tmux_bin: str, + rejected_command: str, +) -> tuple[pathlib.Path, pathlib.Path]: + """Create a wrapper that logs every invocation and fails one subcommand.""" + wrapper = tmp_path / f"fail-{rejected_command}-tmux" + invocation_log = tmp_path / "tmux-invocations" + wrapper.write_text( + "#!/usr/bin/env python3\n" + "import os\n" + "import sys\n" + f"with open({str(invocation_log)!r}, 'a', encoding='utf-8') as stream:\n" + " print(sys.argv[1:], file=stream)\n" + f"if {rejected_command!r} in sys.argv:\n" + " raise SystemExit(1)\n" + f"os.execv({tmux_bin!r}, [{tmux_bin!r}, *sys.argv[1:]])\n", + encoding="utf-8", + ) + wrapper.chmod(0o755) + return wrapper, invocation_log + + +def _stop_after_first_page_wrapper( + tmp_path: pathlib.Path, + tmux_bin: str, + marker: str, +) -> pathlib.Path: + """Build a wrapper that kills the real server partway through a run. + + Trips on the *second* time it sees the between-page identity probe + (``marker``): the first is the baseline captured before any page runs, + the second is the check right after the first page finishes. That is + exactly where a doctest example stopping the lent server -- as + ``with Server()`` does in the excluded ``context_managers.md`` -- would + land, reproduced here without needing to run that page. The probing + command itself is then allowed to reach the now-dead socket and fail. + """ + counter = tmp_path / "probe-count" + wrapper = tmp_path / "stop-after-first-page-tmux" + wrapper.write_text( + "#!/usr/bin/env python3\n" + "import os\n" + "import subprocess\n" + "import sys\n" + f"tmux_bin = {tmux_bin!r}\n" + f"marker = {marker!r}\n" + f"counter = {str(counter)!r}\n" + "args = sys.argv[1:]\n" + "if marker in args:\n" + " seen = 0\n" + " if os.path.exists(counter):\n" + " with open(counter, encoding='utf-8') as fh:\n" + " seen = int(fh.read())\n" + " seen += 1\n" + " with open(counter, 'w', encoding='utf-8') as fh:\n" + " fh.write(str(seen))\n" + " if seen == 2:\n" + " socket_arg = next(a for a in args if a.startswith('-S'))\n" + " subprocess.run([tmux_bin, socket_arg, 'kill-server'])\n" + "os.execv(tmux_bin, [tmux_bin, *args])\n", + encoding="utf-8", + ) + wrapper.chmod(0o755) + return wrapper + + +def test_descriptor_is_the_only_arena_activation_switch() -> None: + """An alias without a descriptor preserves the ordinary doctest path.""" + arena = importlib.import_module("libtmux._arena") + + assert ( + arena.ArenaSpec.from_environ({"LIBTMUX_ARENA_ARTIFACT": "python-exact-binary"}) + is None + ) + assert ( + arena.ArenaSpec.from_environ( + { + "LIBTMUX_ARENA_DESCRIPTOR": "", + "LIBTMUX_ARENA_ARTIFACT": "python-exact-binary", + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "tmux", + } + ) + is None + ) + + +@pytest.mark.parametrize( + "environ", + [ + {"LIBTMUX_ARENA_DESCRIPTOR": "arena"}, + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "", + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "tmux", + }, + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "python-exact-binary", + "LIBTMUX_SOCKET_PATH": "", + "LIBTMUX_TMUX_BIN": "tmux", + }, + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "python-exact-binary", + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "", + }, + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "wrong-artifact", + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "tmux", + }, + ], +) +def test_activated_contract_rejects_empty_or_unknown_values( + environ: dict[str, str], +) -> None: + """An incomplete contract cannot fall through to ambient tmux.""" + arena = importlib.import_module("libtmux._arena") + + with pytest.raises(ValueError): + arena.ArenaSpec.from_environ(environ) + + +@pytest.mark.parametrize("artifact", ["python-exact-binary", "python-workspace-setup"]) +def test_artifact_requires_the_workspace_setup_source( + artifact: str, +) -> None: + """Both audited artifacts bind evidence to the one documented source.""" + arena = importlib.import_module("libtmux._arena") + spec = arena.ArenaSpec.from_environ( + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": artifact, + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "tmux", + } + ) + + assert spec is not None + assert spec.targets_for(pathlib.Path("/repo")) == ( + pathlib.Path("/repo/docs/topics/workspace_setup.md"), + ) + + +def test_an_artifact_may_bind_several_sources( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """One lend can audit more than one page, and every page is required. + + The supervisor takes one evidence record per declared source, so an + artifact that names two pages and runs one is a failure rather than a + partial pass. + """ + arena = importlib.import_module("libtmux._arena") + monkeypatch.setitem( + arena.ARENA_ARTIFACT_TARGETS, + "python-two-pages", + ("docs/topics/workspace_setup.md", "docs/topics/traversal.md"), + ) + spec = arena.ArenaSpec.from_environ( + { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": "python-two-pages", + "LIBTMUX_SOCKET_PATH": "socket", + "LIBTMUX_TMUX_BIN": "tmux", + } + ) + + assert spec is not None + assert spec.targets_for(pathlib.Path("/repo")) == ( + pathlib.Path("/repo/docs/topics/workspace_setup.md"), + pathlib.Path("/repo/docs/topics/traversal.md"), + ) + + +def test_wiring_an_excluded_source_to_an_artifact_is_refused() -> None: + """The static wiring guard is a real check: it can fail, not just decorate. + + ``context_managers.md`` stops the lent server through `with Server()` + (its `Server` name resolves to a factory pinned to the arena socket in + arena mode, so exiting the block runs `kill-server` against the + borrowed daemon). An artifact tuple can never bind it -- this proves + the guard actually rejects such a mapping rather than always agreeing. + """ + arena = importlib.import_module("libtmux._arena") + conflicting = dict(arena.ARENA_ARTIFACT_TARGETS) + conflicting["python-broken"] = (next(iter(arena.ARENA_EXCLUDED_SOURCES)),) + + with pytest.raises(AssertionError): + arena._assert_no_excluded_targets(conflicting) + + +def test_the_committed_artifact_wiring_passes_the_same_guard() -> None: + """Control: today's real ARENA_ARTIFACT_TARGETS never trips the guard.""" + arena = importlib.import_module("libtmux._arena") + + arena._assert_no_excluded_targets(arena.ARENA_ARTIFACT_TARGETS) + + +@pytest.mark.parametrize( + ("artifact", "target"), + [ + ("", "docs/topics/workspace_setup.md"), + ("unknown", "docs/topics/workspace_setup.md"), + ("python-exact-binary", "README.md"), + ], +) +def test_activated_pytest_rejects_invalid_contract_before_talking_to_tmux( + artifact: str, + target: str, + tmp_path: pathlib.Path, +) -> None: + """Bad activation input cannot reach a default or external server.""" + invocation_log = tmp_path / "tmux-invocations" + wrapper = tmp_path / "tmux" + wrapper.write_text( + f"#!/bin/sh\nprintf invoked >> {shlex.quote(str(invocation_log))}\nexit 1\n", + encoding="utf-8", + ) + wrapper.chmod(0o755) + environ = os.environ | { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": artifact, + "LIBTMUX_SOCKET_PATH": "/not-a-tmux-socket", + "LIBTMUX_TMUX_BIN": str(wrapper), + } + + result = _run_arena(environ, "--libtmux-arena-target", target, target) + + assert result.returncode == 4 + assert not invocation_log.exists() + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + + +def test_arena_refuses_a_source_known_to_stop_the_server( + TestServer: t.Callable[..., Server], +) -> None: + """An excluded source is refused by name, not merely absent from a tuple. + + A generic "artifact requires {...}" mismatch would look identical to a + typo from the outside. Requesting the excluded page proves the specific, + named reason fires instead -- before tmux is even touched. + """ + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + excluded = "docs/topics/context_managers.md" + result = _run_arena( + _arena_environ(endpoint, tmux_bin), + "--libtmux-arena-target", + excluded, + excluded, + ) + + assert result.returncode == 4 + combined = result.stdout + result.stderr + assert excluded in combined + arena = importlib.import_module("libtmux._arena") + assert arena.ARENA_EXCLUDED_SOURCES[excluded] in combined + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + _assert_only_hold(endpoint) + + +def test_arena_runs_the_exact_doctest_on_an_external_server( + TestServer: t.Callable[..., Server], + tmp_path: pathlib.Path, +) -> None: + """The selected page owns sessions without taking the external daemon.""" + endpoint = _external_endpoint(TestServer) + assert endpoint.challenge is not None + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + invocation_log = tmp_path / "tmux-invocations" + wrapper = tmp_path / "exact-tmux" + wrapper.write_text( + "#!/bin/sh\n" + f"printf '%s\\n' \"$@\" >> {shlex.quote(str(invocation_log))}\n" + f'exec {shlex.quote(tmux_bin)} "$@"\n', + encoding="utf-8", + ) + wrapper.chmod(0o755) + result = _run_arena( + _arena_environ(endpoint, str(wrapper)), + "--libtmux-arena-target", + TARGET, + TARGET, + ) + + assert result.returncode == 0, result.stdout + result.stderr + evidence_lines = [ + line.removeprefix("LIBTMUX_ARENA_EVIDENCE=") + for line in result.stdout.splitlines() + if line.startswith("LIBTMUX_ARENA_EVIDENCE=") + ] + assert len(evidence_lines) == 1 + evidence = json.loads(evidence_lines[0]) + assert evidence == { + "artifact": "python-exact-binary", + "challenge": endpoint.challenge, + "schema": 1, + "server_pid": int( + endpoint.server.cmd("display-message", "-p", "#{pid}").stdout[0] + ), + "socket_path": endpoint.socket_path, + "source": TARGET, + } + assert invocation_log.read_text(encoding="utf-8") + _assert_only_hold(endpoint) + + +def test_arena_rejects_a_deselected_doctest_subset( + TestServer: t.Callable[..., Server], +) -> None: + """A passing subset cannot produce evidence for the complete source.""" + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + result = _run_arena( + _arena_environ(endpoint, tmux_bin), + "--libtmux-arena-target", + TARGET, + "-k", + "0", + TARGET, + ) + + assert result.returncode == 4 + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + _assert_only_hold(endpoint) + + +def test_arena_rejects_a_deselected_second_source( + TestServer: t.Callable[..., Server], +) -> None: + """A filtered second source cannot disappear from collection proof.""" + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + result = _run_arena( + _arena_environ(endpoint, tmux_bin), + "--libtmux-arena-target", + TARGET, + "-k", + "workspace_setup", + TARGET, + "README.md", + ) + + assert result.returncode == 4 + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + _assert_only_hold(endpoint) + + +def test_arena_rejects_an_external_server_without_a_challenge( + TestServer: t.Callable[..., Server], +) -> None: + """A successful doctest run cannot publish an empty challenge.""" + endpoint = _external_endpoint(TestServer, challenge=None) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + result = _run_arena( + _arena_environ(endpoint, tmux_bin), + "--libtmux-arena-target", + TARGET, + TARGET, + ) + + assert result.returncode != 0 + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + _assert_only_hold(endpoint) + + +def test_arena_rejects_a_wrapper_redirected_socket( + TestServer: t.Callable[..., Server], + tmp_path: pathlib.Path, +) -> None: + """Evidence cannot name a socket different from the requested endpoint.""" + expected = _external_endpoint(TestServer, hold_name="arena-expected-hold") + alternate = _external_endpoint(TestServer, hold_name="arena-alternate-hold") + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + wrapper = tmp_path / "redirect-tmux" + wrapper.write_text( + "#!/usr/bin/env python3\n" + "import os\n" + "import sys\n" + f"alternate_socket = {alternate.socket_path!r}\n" + "arguments = sys.argv[1:]\n" + "for index, argument in enumerate(arguments):\n" + " if argument == '-S':\n" + " arguments[index + 1] = alternate_socket\n" + " elif argument.startswith('-S'):\n" + " arguments[index] = '-S' + alternate_socket\n" + f"os.execv({tmux_bin!r}, [{tmux_bin!r}, *arguments])\n", + encoding="utf-8", + ) + wrapper.chmod(0o755) + result = _run_arena( + _arena_environ(expected, str(wrapper)), + "--libtmux-arena-target", + TARGET, + TARGET, + ) + + assert result.returncode != 0 + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + _assert_only_hold(expected) + _assert_only_hold(alternate) + + +def test_arena_does_not_publish_evidence_after_session_cleanup_fails( + TestServer: t.Callable[..., Server], + tmp_path: pathlib.Path, +) -> None: + """A failed adapter cleanup fails the run instead of hiding a leaked session.""" + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + wrapper, invocation_log = _failing_tmux_wrapper(tmp_path, tmux_bin, "kill-session") + try: + result = _run_arena( + _arena_environ(endpoint, str(wrapper)), + "--libtmux-arena-target", + TARGET, + TARGET, + ) + + assert result.returncode != 0 + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + assert "kill-session" in invocation_log.read_text(encoding="utf-8") + finally: + _remove_adapter_sessions(endpoint) + + _assert_only_hold(endpoint) + + +def test_arena_cleanup_does_not_treat_a_failed_probe_as_an_absent_session( + TestServer: t.Callable[..., Server], + tmp_path: pathlib.Path, +) -> None: + """The adapter cleans its session without a lenient existence probe.""" + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + wrapper, invocation_log = _failing_tmux_wrapper(tmp_path, tmux_bin, "has-session") + try: + result = _run_arena( + _arena_environ(endpoint, str(wrapper)), + "--libtmux-arena-target", + TARGET, + TARGET, + ) + + assert result.returncode == 0, result.stdout + result.stderr + assert "LIBTMUX_ARENA_EVIDENCE=" in result.stdout + assert "has-session" in invocation_log.read_text(encoding="utf-8") + assert [ + session.session_name + for session in endpoint.server.sessions + if session.session_name != endpoint.hold_name + ] == [] + finally: + _remove_adapter_sessions(endpoint) + + +def test_arena_runs_two_pages_and_proves_identity_between_them( + TestServer: t.Callable[..., Server], +) -> None: + """A real multi-page artifact reaps only what it left, one record per page. + + Passing control for the destructive test below. ``self_location.md`` + creates sessions it never kills (``elsewhere``, ``aaa-home``, + ``zzz-guest``) -- the same kind of stray the borrowed six-page run this + artifact is modeled on found (there: ``97``, ``foo``, ``guest``, + ``home``). Teardown must reap them and leave the endpoint's own hold + session untouched, and both evidence records must show the same server + identity. + """ + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + environ = os.environ | { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": TWO_PAGE_ARTIFACT, + "LIBTMUX_SOCKET_PATH": endpoint.socket_path, + "LIBTMUX_TMUX_BIN": tmux_bin, + } + result = _run_arena( + environ, + "--libtmux-arena-target", + FIRST_PAGE, + "--libtmux-arena-target", + SECOND_PAGE, + FIRST_PAGE, + SECOND_PAGE, + ) + + assert result.returncode == 0, result.stdout + result.stderr + evidence = [ + json.loads(line.removeprefix("LIBTMUX_ARENA_EVIDENCE=")) + for line in result.stdout.splitlines() + if line.startswith("LIBTMUX_ARENA_EVIDENCE=") + ] + assert {record["source"] for record in evidence} == {FIRST_PAGE, SECOND_PAGE} + assert len({record["server_pid"] for record in evidence}) == 1 + assert len({record["challenge"] for record in evidence}) == 1 + _assert_only_hold(endpoint) + + +def test_arena_rejects_a_server_that_stops_between_pages( + TestServer: t.Callable[..., Server], + tmp_path: pathlib.Path, +) -> None: + """A server stopped between pages is rejected and names the page. + + Negative test for the between-page identity check: without it, a run + like this would keep going -- and publish evidence -- against whatever + silently answered after the break, exactly the "later call silently + started a replacement with no challenge set" failure mode the excluded + ``context_managers.md`` produces via ``with Server()``. The wrapper + stops the real server the moment the check runs for the first page, + without needing to run that excluded page to prove it. + """ + endpoint = _external_endpoint(TestServer) + tmux_bin = shutil.which("tmux") + assert tmux_bin is not None + wrapper = _stop_after_first_page_wrapper(tmp_path, tmux_bin, ARENA_IDENTITY_PROBE) + environ = os.environ | { + "LIBTMUX_ARENA_DESCRIPTOR": "arena", + "LIBTMUX_ARENA_ARTIFACT": TWO_PAGE_ARTIFACT, + "LIBTMUX_SOCKET_PATH": endpoint.socket_path, + "LIBTMUX_TMUX_BIN": str(wrapper), + } + result = _run_arena( + environ, + "--libtmux-arena-target", + FIRST_PAGE, + "--libtmux-arena-target", + SECOND_PAGE, + FIRST_PAGE, + SECOND_PAGE, + ) + + assert result.returncode != 0 + combined = result.stdout + result.stderr + assert "LIBTMUX_ARENA_EVIDENCE=" not in result.stdout + assert FIRST_PAGE in combined + assert "identity" in combined.lower() + # The original endpoint (hold session included) is gone by design -- the + # wrapper's whole point is simulating that it got stopped. What answers + # the socket afterwards, if anything, is cleaned up by TestServer's own + # finalizer (`_reap_test_server`), which kills by socket name regardless + # of which daemon currently answers there.