From 8ac7126bd0b6187b7c6602c0bc3c2ffa6bb45c83 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Tue, 25 Aug 2026 04:33:27 -0500 Subject: [PATCH 1/3] Tests(feat[arena]): Run doctests in an owned server why: Run audited doctests against an externally owned tmux server without falling back to ambient state or taking daemon ownership. what: - Validate the opt-in arena contract and exact source selection - Isolate each selected doctest in an adapter-owned session - Emit verified endpoint evidence and cover rejection paths --- conftest.py | 206 +++++++++++++++++--- src/libtmux/_arena.py | 42 +++++ tests/test_arena.py | 430 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 655 insertions(+), 23 deletions(-) create mode 100644 src/libtmux/_arena.py create mode 100644 tests/test_arena.py diff --git a/conftest.py b/conftest.py index 88a2656d29..9f84a061ca 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 ArenaSpec from libtmux._internal.control_mode import ControlMode from libtmux.client import Client from libtmux.pane import Pane @@ -25,48 +30,166 @@ 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=" +ARENA_SPEC_KEY: pytest.StashKey[ArenaSpec] = pytest.StashKey() +ARENA_TARGET_KEY: pytest.StashKey[pathlib.Path] = pytest.StashKey() +ARENA_DISCOVERED_PATHS_KEY: pytest.StashKey[frozenset[pathlib.Path]] = pytest.StashKey() +ARENA_DISCOVERED_KEY: pytest.StashKey[frozenset[str]] = pytest.StashKey() +ARENA_COLLECTED_KEY: pytest.StashKey[frozenset[str]] = pytest.StashKey() +ARENA_PASSED_KEY: pytest.StashKey[frozenset[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_target(config: pytest.Config) -> pathlib.Path | None: + """Return the validated arena source for this pytest invocation.""" + return config.stash.get(ARENA_TARGET_KEY, None) + + +def pytest_addoption(parser: pytest.Parser) -> None: + """Register the source selected by the arena adapter.""" + parser.addoption( + "--libtmux-arena-target", + metavar="PATH", + help="Run one audited doctest source against an external tmux server", + ) + + +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_target = config.getoption("libtmux_arena_target") + expected_relative = ArenaSpec.target_for(spec, pathlib.Path()).as_posix() + if raw_target != expected_relative: + msg = f"arena artifact {spec.artifact!r} requires {expected_relative!r}" + raise pytest.UsageError(msg) + root = pathlib.Path(config.rootpath).resolve() + target = spec.target_for(root).resolve(strict=True) + config.stash[ARENA_SPEC_KEY] = spec + config.stash[ARENA_TARGET_KEY] = target + + +def pytest_collection_finish(session: pytest.Session) -> None: + """Reject selections that include anything besides the audited source.""" + target = _arena_target(session.config) + if target 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, frozenset()) + if ( + paths != {target} + or discovered_paths != {target} + or not discovered + or selected != discovered + ): + msg = "arena requires collection of exactly one audited doctest source" + raise pytest.UsageError(msg) + session.config.stash[ARENA_COLLECTED_KEY] = selected + + +def pytest_itemcollected(item: pytest.Item) -> None: + """Record every arena item before pytest applies filters.""" + target = _arena_target(item.config) + if target 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 == target: + discovered = item.config.stash.get(ARENA_DISCOVERED_KEY, frozenset()) + item.config.stash[ARENA_DISCOVERED_KEY] = discovered | {item.nodeid} + + +def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[t.Any]) -> None: + """Record successful arena doctest calls for evidence publication.""" + target = _arena_target(item.config) + if target is None or item.path.resolve() != target: + return + if call.when == "call" and call.excinfo is None: + passed = item.config.stash.get(ARENA_PASSED_KEY, frozenset()) + item.config.stash[ARENA_PASSED_KEY] = passed | {item.nodeid} + @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) + if cleanup.returncode != 0: + msg = "arena session cleanup failed" + raise RuntimeError(msg) @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 +207,40 @@ def setup_session( """Session-level test configuration for pytest.""" if USING_ZSH: request.getfixturevalue("zshrc") + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """Publish evidence only after the selected doctests pass.""" + spec = _arena_spec(session.config) + target = _arena_target(session.config) + if spec is None or target is None or exitstatus != pytest.ExitCode.OK: + return + collected = session.config.stash.get(ARENA_COLLECTED_KEY, frozenset()) + passed = session.config.stash.get(ARENA_PASSED_KEY, frozenset()) + if not collected or passed != collected or session.config.getoption("collectonly"): + return + + server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) + result = server.cmd( + "display-message", + "-p", + "#{pid}\t#{socket_path}\t#{@libtmux_arena_challenge}", + ).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) + evidence = { + "artifact": spec.artifact, + "challenge": parts[2], + "schema": 1, + "server_pid": int(parts[0]), + "socket_path": parts[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..4f96859e3b --- /dev/null +++ b/src/libtmux/_arena.py @@ -0,0 +1,42 @@ +"""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", +} + + +@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 target_for(self, root: pathlib.Path) -> pathlib.Path: + """Resolve the source bound to this artifact inside ``root``.""" + return root / ARENA_ARTIFACT_TARGETS[self.artifact] diff --git a/tests/test_arena.py b/tests/test_arena.py new file mode 100644 index 0000000000..3716178595 --- /dev/null +++ b/tests/test_arena.py @@ -0,0 +1,430 @@ +"""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" + + +@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 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.target_for(pathlib.Path("/repo")) == pathlib.Path( + "/repo/docs/topics/workspace_setup.md" + ) + + +@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_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) + + _assert_only_hold(endpoint) From b38cadc0dfb7f0350ae85749e755f8189e198943 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Fri, 11 Sep 2026 18:45:17 -0500 Subject: [PATCH 2/3] Tests(feat[arena]): Let one artifact audit several pages why: an artifact bound exactly one doctest source, so a page that wanted to share a lent server with its neighbours could not: the supervisor would have to start a server per page. Five pages cost about 421 ms that way against about 85 ms in one lend. The contract on the other side now takes one evidence record per declared source, and this is the side that produces them. what: - `ARENA_ARTIFACT_TARGETS` maps an artifact to a tuple of sources, and `ArenaSpec.targets_for` resolves all of them. - `--libtmux-arena-target` is repeatable, one flag per audited page, and the set passed has to equal the set the artifact declares. - Collection tracks discovered, collected and passed node ids per source, so a declared page that collected nothing is caught rather than averaged away. - One evidence record per source at the end, each naming its own page. Either every requested page gets a record or, on any failure, none do. - A test for the multi-source binding, beside the existing single-source one. The two real artifacts still declare one page each; nothing about what runs today changes. --- conftest.py | 158 ++++++++++++++++++++++++++++-------------- src/libtmux/_arena.py | 10 +-- tests/test_arena.py | 35 +++++++++- 3 files changed, 144 insertions(+), 59 deletions(-) diff --git a/conftest.py b/conftest.py index 9f84a061ca..88b8e4cd61 100644 --- a/conftest.py +++ b/conftest.py @@ -34,11 +34,21 @@ ARENA_EVIDENCE_PREFIX = "LIBTMUX_ARENA_EVIDENCE=" ARENA_SPEC_KEY: pytest.StashKey[ArenaSpec] = pytest.StashKey() -ARENA_TARGET_KEY: pytest.StashKey[pathlib.Path] = 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() -ARENA_DISCOVERED_KEY: pytest.StashKey[frozenset[str]] = pytest.StashKey() -ARENA_COLLECTED_KEY: pytest.StashKey[frozenset[str]] = pytest.StashKey() -ARENA_PASSED_KEY: pytest.StashKey[frozenset[str]] = 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() +) def _arena_spec(config: pytest.Config) -> ArenaSpec | None: @@ -46,17 +56,23 @@ def _arena_spec(config: pytest.Config) -> ArenaSpec | None: return config.stash.get(ARENA_SPEC_KEY, None) -def _arena_target(config: pytest.Config) -> pathlib.Path | None: - """Return the validated arena source for this pytest invocation.""" - return config.stash.get(ARENA_TARGET_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 pytest_addoption(parser: pytest.Parser) -> None: - """Register the source selected by the arena adapter.""" + """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", - help="Run one audited doctest source against an external tmux server", + action="append", + help="Run one audited doctest source against an external tmux server " + "(repeatable)", ) @@ -69,58 +85,69 @@ def pytest_configure(config: pytest.Config) -> None: if spec is None: return - raw_target = config.getoption("libtmux_arena_target") - expected_relative = ArenaSpec.target_for(spec, pathlib.Path()).as_posix() - if raw_target != expected_relative: - msg = f"arena artifact {spec.artifact!r} requires {expected_relative!r}" + raw_targets = frozenset(config.getoption("libtmux_arena_target") or []) + 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() - target = spec.target_for(root).resolve(strict=True) + targets = frozenset(p.resolve(strict=True) for p in spec.targets_for(root)) config.stash[ARENA_SPEC_KEY] = spec - config.stash[ARENA_TARGET_KEY] = target + config.stash[ARENA_TARGETS_KEY] = targets def pytest_collection_finish(session: pytest.Session) -> None: - """Reject selections that include anything besides the audited source.""" - target = _arena_target(session.config) - if target is 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, frozenset()) + discovered = session.config.stash.get(ARENA_DISCOVERED_KEY, {}) + discovered_all = ( + frozenset().union(*discovered.values()) if discovered else frozenset() + ) if ( - paths != {target} - or discovered_paths != {target} - or not discovered - or selected != discovered + 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 one audited doctest source" + msg = "arena requires collection of exactly the audited doctest sources" raise pytest.UsageError(msg) - session.config.stash[ARENA_COLLECTED_KEY] = selected + session.config.stash[ARENA_COLLECTED_KEY] = dict(discovered) def pytest_itemcollected(item: pytest.Item) -> None: """Record every arena item before pytest applies filters.""" - target = _arena_target(item.config) - if target is None: + 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 == target: - discovered = item.config.stash.get(ARENA_DISCOVERED_KEY, frozenset()) - item.config.stash[ARENA_DISCOVERED_KEY] = discovered | {item.nodeid} + 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.""" - target = _arena_target(item.config) - if target is None or item.path.resolve() != target: + 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, frozenset()) - item.config.stash[ARENA_PASSED_KEY] = passed | {item.nodeid} + 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) @@ -176,7 +203,18 @@ def add_doctest_fixtures( finally: if spec is not None: cleanup = server.cmd("kill-session", target=session_name) - if cleanup.returncode != 0: + # 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) @@ -210,14 +248,28 @@ def setup_session( def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: - """Publish evidence only after the selected doctests pass.""" + """Publish one evidence record per audited source, after all of them pass. + + fail-closed is preserved 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) - target = _arena_target(session.config) - if spec is None or target is None or exitstatus != pytest.ExitCode.OK: + targets = _arena_targets(session.config) + if spec is None or targets is None or exitstatus != pytest.ExitCode.OK: return - collected = session.config.stash.get(ARENA_COLLECTED_KEY, frozenset()) - passed = session.config.stash.get(ARENA_PASSED_KEY, frozenset()) - if not collected or passed != collected or session.config.getoption("collectonly"): + collected_by_target = session.config.stash.get(ARENA_COLLECTED_KEY, {}) + passed_by_target = session.config.stash.get(ARENA_PASSED_KEY, {}) + if ( + not collected_by_target + or collected_by_target.keys() != targets + or passed_by_target != collected_by_target + or session.config.getoption("collectonly") + ): return server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) @@ -233,14 +285,16 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: 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) - evidence = { - "artifact": spec.artifact, - "challenge": parts[2], - "schema": 1, - "server_pid": int(parts[0]), - "socket_path": parts[1], - "source": target.relative_to(session.config.rootpath).as_posix(), - } - print( - "\n" + ARENA_EVIDENCE_PREFIX + json.dumps(evidence, sort_keys=True), flush=True - ) + for target in sorted(targets, key=lambda p: p.as_posix()): + evidence = { + "artifact": spec.artifact, + "challenge": parts[2], + "schema": 1, + "server_pid": int(parts[0]), + "socket_path": parts[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 index 4f96859e3b..9da807c263 100644 --- a/src/libtmux/_arena.py +++ b/src/libtmux/_arena.py @@ -7,8 +7,8 @@ import typing as t ARENA_ARTIFACT_TARGETS = { - "python-exact-binary": "docs/topics/workspace_setup.md", - "python-workspace-setup": "docs/topics/workspace_setup.md", + "python-exact-binary": ("docs/topics/workspace_setup.md",), + "python-workspace-setup": ("docs/topics/workspace_setup.md",), } @@ -37,6 +37,6 @@ def from_environ(cls, environ: t.Mapping[str, str]) -> ArenaSpec | None: raise ValueError(msg) return cls(artifact=artifact, socket_path=socket_path, tmux_bin=tmux_bin) - def target_for(self, root: pathlib.Path) -> pathlib.Path: - """Resolve the source bound to this artifact inside ``root``.""" - return root / ARENA_ARTIFACT_TARGETS[self.artifact] + 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 index 3716178595..217ede263a 100644 --- a/tests/test_arena.py +++ b/tests/test_arena.py @@ -186,8 +186,39 @@ def test_artifact_requires_the_workspace_setup_source( ) assert spec is not None - assert spec.target_for(pathlib.Path("/repo")) == pathlib.Path( - "/repo/docs/topics/workspace_setup.md" + 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"), ) From ae442fd8fee1ba8fe290ad1da8a9c01558983e5d Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Fri, 11 Sep 2026 20:21:59 -0500 Subject: [PATCH 3/3] Tests(feat[arena]): Audit two documentation pages on one lent server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit why: an artifact could bind several pages but none did, so the capability was unexercised and the cost it exists to avoid — one server per page — was still being paid. Sharing a server also raises the question the single-page gate never had to answer: whether the server a page finishes on is the one it started on. what: - `python-workspace-and-location` audits `workspace_setup.md` and `self_location.md` together. Both take the server the fixture hands them. - `context_managers.md` is excluded by name and reason, not by omission: every example there opens `with Server()`, which under the arena resolves to the lent socket, so leaving the block stops the borrowed server. An import-time guard refuses to load if any artifact ever names an excluded page, and the request is rejected before tmux is touched. - The server's identity — pid, socket and challenge — is proved after each page, not only at the end, and a change names the page it happened after. - Teardown reaps only the sessions the run created, diffed against a baseline taken before it, and only while the identity still matches. It never stops the server. - A destructive test reproduces the failure this is for: a wrapper that stops the server between pages makes the run exit nonzero with no evidence, naming the page. Its control runs the two pages and proves identity between them. --- conftest.py | 195 ++++++++++++++++++++++++++++++++++++++---- src/libtmux/_arena.py | 47 ++++++++++ tests/test_arena.py | 191 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 414 insertions(+), 19 deletions(-) diff --git a/conftest.py b/conftest.py index 88b8e4cd61..c8190aeb0e 100644 --- a/conftest.py +++ b/conftest.py @@ -21,7 +21,7 @@ import pytest from _pytest.doctest import DoctestItem -from libtmux._arena import ArenaSpec +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 @@ -33,6 +33,11 @@ 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. @@ -49,6 +54,16 @@ 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: @@ -61,6 +76,75 @@ def _arena_targets(config: pytest.Config) -> frozenset[pathlib.Path] | None: 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. @@ -86,6 +170,22 @@ def pytest_configure(config: pytest.Config) -> 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()) ) @@ -97,6 +197,20 @@ def pytest_configure(config: pytest.Config) -> None: 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.""" @@ -121,6 +235,17 @@ def pytest_collection_finish(session: pytest.Session) -> None: 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.""" @@ -218,6 +343,26 @@ def add_doctest_fixtures( 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( @@ -248,9 +393,17 @@ def setup_session( def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: - """Publish one evidence record per audited source, after all of them pass. + """Reap the run's stray sessions, then publish evidence once all pass. - fail-closed is preserved because ``exitstatus`` is only + 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 @@ -260,38 +413,42 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: """ spec = _arena_spec(session.config) targets = _arena_targets(session.config) - if spec is None or targets is None or exitstatus != pytest.ExitCode.OK: + 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 ( - not collected_by_target + 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 - server = Server(socket_path=spec.socket_path, tmux_bin=spec.tmux_bin) - result = server.cmd( - "display-message", - "-p", - "#{pid}\t#{socket_path}\t#{@libtmux_arena_challenge}", - ).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]: + 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": parts[2], + "challenge": baseline_identity[2], "schema": 1, - "server_pid": int(parts[0]), - "socket_path": parts[1], + "server_pid": baseline_identity[0], + "socket_path": baseline_identity[1], "source": target.relative_to(session.config.rootpath).as_posix(), } print( diff --git a/src/libtmux/_arena.py b/src/libtmux/_arena.py index 9da807c263..2dc3895ec2 100644 --- a/src/libtmux/_arena.py +++ b/src/libtmux/_arena.py @@ -9,8 +9,55 @@ 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: diff --git a/tests/test_arena.py b/tests/test_arena.py index 217ede263a..455ebb7eee 100644 --- a/tests/test_arena.py +++ b/tests/test_arena.py @@ -19,6 +19,12 @@ 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) @@ -109,6 +115,50 @@ def _failing_tmux_wrapper( 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") @@ -222,6 +272,30 @@ def test_an_artifact_may_bind_several_sources( ) +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"), [ @@ -257,6 +331,35 @@ def test_activated_pytest_rejects_invalid_contract_before_talking_to_tmux( 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, @@ -458,4 +561,92 @@ def test_arena_cleanup_does_not_treat_a_failed_probe_as_an_absent_session( 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.