"""pytest fixtures for gammaray-mcp bridge tests. All paths are configurable via environment variables. run_tests.sh is the canonical configuration point — it sets sensible defaults and exports them before calling pytest. If you prefer to run pytest directly, set these: GAMMARAY_BIN — path to the gammaray probe binary GAMMARAY_LIB_DIR — directory containing GammaRay shared libraries QML_RUNNER — path to the QML runtime executable TEST_QML — path to the test QML application BRIDGE_EXE — path to the bridge executable BRIDGE_RUN_SCRIPT — path to bridge/run.sh """ import os import subprocess import time from pathlib import Path import pytest from mcp_client import McpClient PROJECT_ROOT = Path(__file__).resolve().parent.parent # All paths read from env vars, with sensible project-local defaults. # Users only need to override these when GammaRay is installed elsewhere. GAMMARAY_BIN = Path(os.environ.get( "GAMMARAY_BIN", str(PROJECT_ROOT / "install-prefix" / "bin" / "gammaray"), )) GAMMARAY_LIB_DIR = os.environ.get( "GAMMARAY_LIB_DIR", str(PROJECT_ROOT / "install-prefix" / "lib"), ) QML_RUNNER = Path(os.environ.get( "QML_RUNNER", "/usr/lib/qt6/bin/qml", )) TEST_QML = Path(os.environ.get( "TEST_QML", str(PROJECT_ROOT / "tests" / "testapp" / "main.qml"), )) BRIDGE_EXE = Path(os.environ.get( "BRIDGE_EXE", str(PROJECT_ROOT / "bridge" / "build" / "qml-sg-mcp-bridge"), )) BRIDGE_RUN_SCRIPT = Path(os.environ.get( "BRIDGE_RUN_SCRIPT", str(PROJECT_ROOT / "bridge" / "run.sh"), )) def pytest_addoption(parser): parser.addoption( "--no-probe", action="store_true", default=False, help="Skip tests that need a running probe", ) parser.addoption( "--probe-timeout", type=int, default=15, help="Seconds to wait for probe to be ready", ) def pytest_collection_modifyitems(config, items): if config.getoption("--no-probe"): skip_probe = pytest.mark.skip(reason="Skipping probe-dependent tests (--no-probe)") for item in items: if "probe" in item.keywords: item.add_marker(skip_probe) @pytest.fixture(scope="session") def bridge_path() -> Path: if not BRIDGE_EXE.exists(): pytest.skip(f"Bridge not built at {BRIDGE_EXE}") return BRIDGE_EXE @pytest.fixture(scope="session") def run_script_path() -> Path: if not BRIDGE_RUN_SCRIPT.exists(): pytest.skip(f"run.sh not found at {BRIDGE_RUN_SCRIPT}") return BRIDGE_RUN_SCRIPT @pytest.fixture(scope="module") def probe_process(request) -> subprocess.Popen | None: """Start a GammaRay probe process (session-scoped, one per test run).""" if request.config.getoption("--no-probe"): return None if not GAMMARAY_BIN.exists(): pytest.skip(f"GammaRay probe binary not found at {GAMMARAY_BIN}") if not TEST_QML.exists(): pytest.skip(f"Test QML file not found at {TEST_QML}") if not QML_RUNNER.exists(): pytest.skip(f"QML runner not found at {QML_RUNNER}") env = os.environ.copy() env["QT_QPA_PLATFORM"] = "offscreen" env["LD_LIBRARY_PATH"] = GAMMARAY_LIB_DIR proc = None used_systemd = False # Try systemd-run first for clean detachment subprocess.run( ["systemctl", "--user", "stop", "gammaray-probe-test.service"], capture_output=True, timeout=5, ) subprocess.run( ["systemctl", "--user", "reset-failed", "gammaray-probe-test.service"], capture_output=True, timeout=5, ) result = subprocess.run( [ "systemd-run", "--user", "--no-block", "--unit=gammaray-probe-test", "--same-dir", "--working-directory", str(TEST_QML.parent), "-E", f"QT_QPA_PLATFORM={env['QT_QPA_PLATFORM']}", "-E", f"LD_LIBRARY_PATH={env['LD_LIBRARY_PATH']}", str(GAMMARAY_BIN), "--inject-only", "--listen", "tcp://127.0.0.1:11732", "--injector", "preload", str(QML_RUNNER), str(TEST_QML), ], capture_output=True, text=True, timeout=10, ) if result.returncode == 0: used_systemd = True proc = subprocess.Popen(["true"]) # sentinel: probe is external else: # Fallback: direct spawn proc = subprocess.Popen( [str(GAMMARAY_BIN), "--inject-only", "--listen", "tcp://127.0.0.1:11732", "--injector", "preload", str(QML_RUNNER), str(TEST_QML)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def cleanup(): if proc: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() if used_systemd: subprocess.run( ["systemctl", "--user", "stop", "gammaray-probe-test.service"], capture_output=True, timeout=5, ) subprocess.run( ["systemctl", "--user", "reset-failed", "gammaray-probe-test.service"], capture_output=True, timeout=5, ) request.addfinalizer(cleanup) # Wait for probe to be ready timeout = request.config.getoption("--probe-timeout") deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: c = McpClient(bridge_path=BRIDGE_EXE) c.initialize() result = c.call_tool("connectProbeDefault") c.close() if isinstance(result, dict) and result.get("connected"): break except Exception: pass time.sleep(1) else: pytest.fail(f"Probe did not become ready within {timeout}s") return proc @pytest.fixture def bridge(run_script_path) -> McpClient: """Start a fresh bridge instance per test.""" c = McpClient(bridge_path=run_script_path) c.initialize() yield c # Gracefully disconnect from the probe before killing the bridge. # GammaRay's QTcpSocket::disconnectFromHost() is asynchronous — the # Endpoint::disconnected signal (which clears m_socket) fires on a later # event loop iteration. If we kill the bridge (SIGTERM) while m_socket is # still set, the probe does not immediately detect the disconnection, and # the next test's bridge cannot connect. Sending disconnectProbe() first # triggers a clean GammaRay protocol-level shutdown that waits for the # async disconnect to complete, releasing the probe for the next test. try: c.call_tool("disconnectProbe", timeout=5) except Exception: pass # bridge may have already crashed; best-effort c.close() # Allow the probe to detect the disconnection before the next test # connects. time.sleep(1) @pytest.fixture def connected_bridge(bridge, probe_process) -> McpClient: """Bridge connected to a running probe (retries on transient failures).""" if probe_process is None: pytest.skip("No probe available (try --no-probe to skip probe-dependent tests)") deadline = time.monotonic() + 30 last_error = None while time.monotonic() < deadline: result = bridge.call_tool("connectProbeDefault") if isinstance(result, dict) and result.get("connected"): return bridge last_error = result time.sleep(3) raise AssertionError(f"Failed to connect to probe after 30s: {last_error}")