gammaray-mcp/tests/conftest.py

"""pytest fixtures for gammaray-mcp bridge tests."""

import os
import subprocess
import time
from pathlib import Path

import pytest

from mcp_client import McpClient

PROJECT_ROOT = Path(__file__).resolve().parent.parent
BRIDGE_EXE = PROJECT_ROOT / "bridge" / "build" / "qml-sg-mcp-bridge"
RUN_SCRIPT = PROJECT_ROOT / "bridge" / "run.sh"
PROBE_BIN = PROJECT_ROOT / "install-prefix" / "bin" / "gammaray"
TEST_QML = Path(os.environ.get("TEST_QML", str(Path.home() / "Sources" / "testqml" / "relayout.qml")))
QML_RUNNER = "/usr/lib/qt6/bin/qml"


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:
    """Path to the bridge executable."""
    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:
    """Path to the run.sh wrapper."""
    if not RUN_SCRIPT.exists():
        pytest.skip(f"run.sh not found at {RUN_SCRIPT}")
    return RUN_SCRIPT


@pytest.fixture(scope="session")
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 PROBE_BIN.exists():
        pytest.skip(f"GammaRay probe binary not found at {PROBE_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"] = str(PROJECT_ROOT / "install-prefix" / "lib")

    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(PROBE_BIN),
            "--inject-only", "--listen", "tcp://127.0.0.1:11732",
            "--injector", "preload",
            QML_RUNNER, str(TEST_QML),
        ],
        capture_output=True, text=True, timeout=10,
    )
    if result.returncode == 0:
        used_systemd = True
    else:
        # Fallback: direct spawn
        proc = subprocess.Popen(
            [str(PROBE_BIN), "--inject-only", "--listen", "tcp://127.0.0.1:11732",
             "--injector", "preload", 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
    c.close()


@pytest.fixture
def connected_bridge(bridge, probe_process) -> McpClient:
    """Bridge connected to a running probe."""
    if probe_process is None:
        pytest.skip("No probe available")
    result = bridge.call_tool("connectProbeDefault")
    assert isinstance(result, dict), f"connectProbeDefault failed: {result}"
    assert result.get("connected"), f"Not connected: {result}"
    return bridge