blumia/gammaray-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

make test run faster

Browse Source
This commit is contained in:
Wang Zichong
2026-07-08 10:46:14 +08:00
parent 268483577e
commit 591e786a0d
6 changed files with 104 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
PLAN.md
View File

@@ -413,7 +413,7 @@ The fix was NOT a code change but a sequencing fix in `ensureNodeSelected()`:
All items from the original blocker resolved. Remaining minor issues:
- `listScenegraphNodes` first-call fill stability — first call may show "Loading..." for deep nodes (RemoteModel lazy-fetch behavior); second call returns the full tree. The `primeAndWait()` loop handles this transparently, but single-call would be nicer.
- `walkItemChildren` `flagsToJson` uses hardcoded flag bit values (1,2,4,8,16,32,64) — fragile if GammaRay's `ItemFlags` enum changes. Should read from `common/quickitemmodelroles.h` if it were installed.
- Full test suite takes ~8 minutes due to probe start/stop overhead and 500-1500ms settle delays per tool call. Could be optimized with module-scoped bridge fixtures and shared probe process.
- Full test suite ~3.5 min (session-scoped probe + module-scoped bridge + 0.5s settle delays). Connection tests still get fresh bridges per test (function-scoped).
## Step 7: Launch and run

99
tests/conftest.py
View File

@@ -13,6 +13,7 @@ before calling pytest. If you prefer to run pytest directly, set these:
"""
import os
import socket
import subprocess
import time
from pathlib import Path
@@ -24,7 +25,6 @@ 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"),
@@ -88,9 +88,22 @@ def run_script_path() -> Path:
return BRIDGE_RUN_SCRIPT
@pytest.fixture(scope="module")
def _probe_accepting_connections(timeout: int = 20) -> bool:
"""Check if the probe is accepting connections via raw TCP."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
s = socket.create_connection(("127.0.0.1", 11732), timeout=2)
s.close()
return True
except (OSError, socket.timeout):
time.sleep(1)
return False
@pytest.fixture(scope="session")
def probe_process(request) -> subprocess.Popen | None:
"""Start a GammaRay probe process (session-scoped, one per test run)."""
"""Start a GammaRay probe process (session-scoped, one per entire test run)."""
if request.config.getoption("--no-probe"):
return None
@@ -105,10 +118,6 @@ def probe_process(request) -> subprocess.Popen | None:
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,
@@ -133,83 +142,51 @@ def probe_process(request) -> subprocess.Popen | None:
],
capture_output=True, text=True, timeout=10,
)
if result.returncode == 0:
used_systemd = True
proc = subprocess.Popen(["true"]) # sentinel: probe is external
else:
timeout = request.config.getoption("--probe-timeout")
if not _probe_accepting_connections(timeout):
# 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,
)
if not _probe_accepting_connections(timeout):
pytest.fail(f"Probe did not become ready within {timeout}s")
else:
proc = subprocess.Popen(["true"])
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,
)
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
@pytest.fixture(scope="module")
def bridge(run_script_path) -> McpClient:
"""Start a fresh bridge instance per test."""
"""Start a bridge instance per module, reused across tests in the module."""
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
pass
c.close()
# Allow the probe to detect the disconnection before the next test
# connects.
time.sleep(1)
time.sleep(0.2)
@pytest.fixture
@pytest.fixture(scope="module")
def connected_bridge(bridge, probe_process) -> McpClient:
"""Bridge connected to a running probe (retries on transient failures)."""
"""Bridge connected to a running probe (module-scoped, reused across tests)."""
if probe_process is None:
pytest.skip("No probe available (try --no-probe to skip probe-dependent tests)")
deadline = time.monotonic() + 30
@@ -219,5 +196,5 @@ def connected_bridge(bridge, probe_process) -> McpClient:
if isinstance(result, dict) and result.get("connected"):
return bridge
last_error = result
time.sleep(3)
time.sleep(1)
raise AssertionError(f"Failed to connect to probe after 30s: {last_error}")

86
tests/test_connection.py
View File

@@ -1,67 +1,89 @@
"""Tests for connection management tools."""
"""Tests for connection management tools.
These tests need function-scoped bridges because they explicitly test
connect/disconnect/reconnect lifecycle. The conftest.py module-scoped
fixtures would not work here since they share a single connection.
"""
import time
import pytest
from mcp_client import McpClient
@pytest.fixture(scope="function")
def local_bridge(run_script_path) -> McpClient:
c = McpClient(bridge_path=run_script_path)
c.initialize()
yield c
try:
c.call_tool("disconnectProbe", timeout=5)
except Exception:
pass
c.close()
time.sleep(0.2)
@pytest.fixture(scope="function")
def local_connected_bridge(local_bridge, probe_process) -> McpClient:
if probe_process is None:
pytest.skip("No probe available")
deadline = time.monotonic() + 30
last_error = None
while time.monotonic() < deadline:
result = local_bridge.call_tool("connectProbeDefault")
if isinstance(result, dict) and result.get("connected"):
return local_bridge
last_error = result
time.sleep(1)
raise AssertionError(f"Failed to connect to probe after 30s: {last_error}")
class TestConnection:
"""Verify connectProbe, probeStatus, disconnectProbe work."""
def test_tools_listed(self, bridge):
tools = bridge.list_tools()
def test_tools_listed(self, local_bridge):
tools = local_bridge.list_tools()
names = [t["name"] for t in tools]
for expected in ["connectProbe", "connectProbeDefault",
"disconnectProbe", "probeStatus"]:
assert expected in names, f"Missing tool: {expected}"
def test_probe_status_disconnected(self, bridge):
status = bridge.call_tool("probeStatus")
def test_probe_status_disconnected(self, local_bridge):
status = local_bridge.call_tool("probeStatus")
assert isinstance(status, dict)
assert status.get("state") in ("disconnected", "connecting")
assert status.get("ready") is False
def test_connect_and_disconnect(self, bridge, probe_process):
def test_connect_and_disconnect(self, local_bridge, probe_process):
if probe_process is None:
pytest.skip("No probe available")
# Connect
result = bridge.call_tool("connectProbeDefault")
result = local_bridge.call_tool("connectProbeDefault")
assert isinstance(result, dict), f"connectProbeDefault failed: {result}"
assert result.get("connected"), f"Not connected: {result}"
# Verify status
status = bridge.call_tool("probeStatus")
status = local_bridge.call_tool("probeStatus")
assert isinstance(status, dict)
assert status.get("state") == "ready"
assert status.get("ready") is True
assert "tcp://" in status.get("url", "")
# Now disconnect
result = bridge.call_tool("disconnectProbe")
result = local_bridge.call_tool("disconnectProbe")
assert isinstance(result, dict)
assert result.get("state") == "disconnected"
def test_reconnect(self, bridge, probe_process):
def test_reconnect(self, local_bridge, probe_process):
if probe_process is None:
pytest.skip("No probe available")
bridge.call_tool("connectProbeDefault")
# Disconnect
bridge.call_tool("disconnectProbe")
# Reconnect with explicit args
result = bridge.call_tool("connectProbe",
{"host": "127.0.0.1", "port": 11732})
local_bridge.call_tool("connectProbeDefault")
local_bridge.call_tool("disconnectProbe")
result = local_bridge.call_tool("connectProbe",
{"host": "127.0.0.1", "port": 11732})
assert isinstance(result, dict)
assert result.get("connected") is True
assert result.get("state") == "ready"
def test_connect_default_after_disconnect(self, bridge, probe_process):
if probe_process is None:
pytest.skip("No probe available")
bridge.call_tool("connectProbeDefault")
bridge.call_tool("disconnectProbe")
result = bridge.call_tool("connectProbeDefault")
assert isinstance(result, dict)
assert result.get("connected") is True
assert result.get("state") == "ready"
def test_connect_default_after_disconnect(self, connected_bridge):
connected_bridge.call_tool("disconnectProbe")
result = connected_bridge.call_tool("connectProbeDefault")
def test_reconnect_default_after_disconnect(self, local_connected_bridge):
local_connected_bridge.call_tool("disconnectProbe")
result = local_connected_bridge.call_tool("connectProbeDefault")
assert isinstance(result, dict)
assert result.get("connected") is True

11
tests/test_items.py
View File

@@ -1,7 +1,6 @@
"""Tests for QML item selection and property inspection."""
import pytest
import time
class TestItemProperties:
@@ -18,7 +17,7 @@ class TestItemProperties:
def test_select_quick_item(self, connected_bridge):
# First get items
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
assert isinstance(items, list) and len(items) > 0, "No items found"
@@ -49,7 +48,7 @@ class TestItemProperties:
@pytest.mark.probe
def test_get_item_properties(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
def find_any_item(items_list):
@@ -74,7 +73,7 @@ class TestItemProperties:
def test_item_has_position(self, connected_bridge):
"""Verify that a Button item has x, y, width, height properties."""
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
# Find a Button_QMLTYPE item
@@ -102,7 +101,7 @@ class TestItemProperties:
def test_item_has_visual_properties(self, connected_bridge):
"""Verify items have opacity, visible, z, rotation, scale."""
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
def find_addr(items_list):
@@ -128,7 +127,7 @@ class TestItemProperties:
def test_property_model_has_groups(self, connected_bridge):
"""Verify items have nested property groups (background, contentItem, etc.)."""
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
def find_button(items_list):

15
tests/test_navigation.py
View File

@@ -30,23 +30,16 @@ class TestNavigation:
@pytest.mark.probe
def test_list_items(self, connected_bridge):
# Select window first so item model populates
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
import time
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
assert isinstance(items, list)
# May be empty if model hasn't populated yet; that's OK
if items:
assert isinstance(items[0], dict)
assert "type" in items[0]
@pytest.mark.probe
def test_list_items_with_window_count(self, connected_bridge):
"""Collect all items and verify at least some are found."""
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
import time
time.sleep(1)
items = connected_bridge.call_tool("listQuickItems")
def count_all(items_list):
@@ -63,21 +56,15 @@ class TestNavigation:
@pytest.mark.probe
def test_list_scenegraph_nodes(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
import time
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
assert isinstance(nodes, list)
if nodes:
assert isinstance(nodes[0], dict)
# Root should be "Root Node"
assert nodes[0].get("type") == "Root Node"
@pytest.mark.probe
def test_scenegraph_has_geometry_nodes(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
import time
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
def find_geometry(items_list):
@@ -90,4 +77,4 @@ class TestNavigation:
return None
geo = find_geometry(nodes)
assert geo is not None, "No Geometry Node found in SG tree"
assert geo is not None, "No Geometry Node found in SG tree"

11
tests/test_scenegraph.py
View File

@@ -1,7 +1,6 @@
"""Tests for SceneGraph node selection and geometry/material tools."""
import pytest
import time
class TestSceneGraph:
@@ -21,7 +20,7 @@ class TestSceneGraph:
@pytest.mark.probe
def test_select_scenegraph_node(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
# Find first Geometry Node
@@ -47,7 +46,7 @@ class TestSceneGraph:
@pytest.mark.probe
def test_get_node_vertices(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
def find_geo(items_list):
@@ -71,7 +70,7 @@ class TestSceneGraph:
@pytest.mark.probe
def test_get_node_adjacency(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
def find_geo(items_list):
@@ -94,7 +93,7 @@ class TestSceneGraph:
@pytest.mark.probe
def test_get_material_shaders(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
def find_geo(items_list):
@@ -118,7 +117,7 @@ class TestSceneGraph:
@pytest.mark.probe
def test_get_material_properties(self, connected_bridge):
connected_bridge.call_tool("selectQuickWindow", {"index": 0})
time.sleep(1)
nodes = connected_bridge.call_tool("listScenegraphNodes")
def find_geo(items_list):