diff --git a/test/conftest.py b/test/conftest.py index 01adbcb1..21e22c57 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,7 +1,10 @@ from __future__ import annotations +import glob +import logging import os import stat +import subprocess import sys from pathlib import Path @@ -81,3 +84,63 @@ def _install_provider_stubs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> monkeypatch.setenv("STUB_DELAY", "1.5") monkeypatch.setenv("CCB_REPLY_LANG", "en") monkeypatch.setenv("CCB_CLAUDE_SKILLS", "0") + + +# ============================================================ +# CCB tmux daemon leak cleanup. +# Tests in test_v2_phase2_entrypoint.py spawn `ccb` via subprocess. +# Those subprocesses start the CCB keeper which wraps itself in an +# independent tmux daemon (socket under tmp_path/.ccb/ccbd/tmux.sock +# or /run/user/$UID/ccb-runtime/tmux-*.sock). The test's in-process +# app.shutdown() cannot reach those daemons. +# ============================================================ + +_leak_logger = logging.getLogger(__name__) + + +def _safe_kill_tmux_server(sock: str) -> None: + try: + result = subprocess.run( + ['tmux', '-S', sock, 'kill-server'], + timeout=5, check=False, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + if result.returncode != 0: + _leak_logger.warning( + "cleanup: tmux kill-server %s returned rc=%d", sock, result.returncode + ) + except FileNotFoundError: + return + except subprocess.TimeoutExpired: + _leak_logger.warning("cleanup: tmux kill-server %s timed out after 5s", sock) + + +def _runtime_socket_pattern() -> str: + return f'/run/user/{os.getuid()}/ccb-runtime/tmux-*.sock' + + +@pytest.fixture(autouse=True) +def _cleanup_ccb_tmux_per_test(tmp_path): + before = set(glob.glob(_runtime_socket_pattern())) + try: + yield + finally: + for sock_path in tmp_path.rglob('.ccb/ccbd/tmux.sock'): + _safe_kill_tmux_server(str(sock_path)) + after = set(glob.glob(_runtime_socket_pattern())) + for sock in after - before: + _safe_kill_tmux_server(sock) + + +@pytest.fixture(autouse=True, scope='session') +def _cleanup_ccb_tmux_session_end(tmp_path_factory): + before = set(glob.glob(_runtime_socket_pattern())) + try: + yield + finally: + base_tmp = tmp_path_factory.getbasetemp() + for sock_path in base_tmp.rglob('.ccb/ccbd/tmux.sock'): + _safe_kill_tmux_server(str(sock_path)) + after = set(glob.glob(_runtime_socket_pattern())) + for sock in after - before: + _safe_kill_tmux_server(sock)