diff --git a/PLAN.md b/PLAN.md index d4f7f6c..ec3d1d3 100644 --- a/PLAN.md +++ b/PLAN.md @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py index 2265a10..080fd84 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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}") \ No newline at end of file diff --git a/tests/test_connection.py b/tests/test_connection.py index ea06824..6673bd3 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -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 \ No newline at end of file diff --git a/tests/test_items.py b/tests/test_items.py index 1479de5..2f75b02 100644 --- a/tests/test_items.py +++ b/tests/test_items.py @@ -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): diff --git a/tests/test_navigation.py b/tests/test_navigation.py index 0e691a7..c342f65 100644 --- a/tests/test_navigation.py +++ b/tests/test_navigation.py @@ -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" \ No newline at end of file + assert geo is not None, "No Geometry Node found in SG tree" diff --git a/tests/test_scenegraph.py b/tests/test_scenegraph.py index 77dd24c..de55a1a 100644 --- a/tests/test_scenegraph.py +++ b/tests/test_scenegraph.py @@ -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):