mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-28 16:11:27 +01:00
test(gui): isolate window remapping from BEC and RPC dispatch
This commit is contained in:
1 parent
74f9cf0ff1
commit
44a42bbfa2
3 files changed
+145
-1
No files matched your search
@@ -141,6 +141,19 @@ For an XWayland comparison on a Linux desktop with XWayland available, start a f
|
||||
`QT_QPA_PLATFORM=xcb bec --session flomni` and verify that `qt_platform` is `xcb`. This sets the backend
|
||||
for that invocation only. For plugin-loading details, start with `QT_DEBUG_PLUGINS=1 bec --session flomni`.
|
||||
|
||||
To isolate a visibility problem from BEC, run the plain Qt probe directly from this checkout,
|
||||
using the same Python environment as the affected GUI:
|
||||
|
||||
```bash
|
||||
QT_QPA_PLATFORM=wayland python bec_widgets/examples/qt_window_probe.py
|
||||
```
|
||||
|
||||
Type `cycle` repeatedly in its terminal to test plain Qt hide/show. Compare with separate `hide`
|
||||
and `show` commands; `raise` tests activation without hiding. The probe reports Qt's state but
|
||||
cannot determine whether another application's window obscures it. Record what you see alongside
|
||||
the output and the result of `gnome-shell --version` when reporting a GNOME issue. Run again with
|
||||
`QT_QPA_PLATFORM=xcb` to compare backends. This script starts no BEC services and imports no BEC code.
|
||||
|
||||
### 4. Rapid development (extensible by design)
|
||||
|
||||
Build new widgets fast: Inherit from `BECWidget`, list your RPC methods in `USER_ACCESS`, and use `bec_dispatcher` to
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Probe window visibility from a terminal using only Qt, without starting BEC.
|
||||
|
||||
Run this file directly with the affected environment's Python interpreter.
|
||||
Enter ``cycle`` repeatedly to hide/show a plain QWidget in one event-loop callback.
|
||||
Compare with separate ``hide`` and ``show`` commands and with ``raise``.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
from qtpy import API_NAME
|
||||
from qtpy.QtCore import QSocketNotifier, QTimer, qVersion
|
||||
from qtpy.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run a plain Qt window controlled through standard input on macOS or Linux."""
|
||||
app = QApplication(sys.argv)
|
||||
window = QWidget()
|
||||
window.setWindowTitle("Qt window visibility probe")
|
||||
window.resize(800, 600)
|
||||
layout = QVBoxLayout(window)
|
||||
layout.addWidget(QLabel("Control this plain Qt window from the terminal."))
|
||||
window.show()
|
||||
handle = window.windowHandle()
|
||||
command_number = 0
|
||||
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"pid": os.getpid(),
|
||||
"python": sys.executable,
|
||||
"qt_binding": API_NAME,
|
||||
"qt_version": qVersion(),
|
||||
"qt_platform": app.platformName(),
|
||||
"session_type": os.environ.get("XDG_SESSION_TYPE"),
|
||||
"desktop": os.environ.get("XDG_CURRENT_DESKTOP"),
|
||||
}
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
print("Commands: cycle, hide, show, raise, state, quit", flush=True)
|
||||
|
||||
def report(stage: str, number: int) -> None:
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"command": number,
|
||||
"stage": stage,
|
||||
"visible": window.isVisible(),
|
||||
"native_visible": handle.isVisible(),
|
||||
"active": window.isActiveWindow(),
|
||||
"exposed": handle.isExposed(),
|
||||
"minimized": window.isMinimized(),
|
||||
}
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
|
||||
notifier = QSocketNotifier(sys.stdin.fileno(), QSocketNotifier.Type.Read, window)
|
||||
|
||||
def read_command() -> None:
|
||||
nonlocal command_number
|
||||
line = sys.stdin.readline()
|
||||
if not line or line.strip() == "quit":
|
||||
notifier.setEnabled(False)
|
||||
app.quit()
|
||||
return
|
||||
command = line.strip()
|
||||
command_number += 1
|
||||
number = command_number
|
||||
report(f"before {command}", number)
|
||||
if command == "cycle":
|
||||
window.hide()
|
||||
window.show()
|
||||
elif command == "hide":
|
||||
window.hide()
|
||||
elif command == "show":
|
||||
window.show()
|
||||
elif command == "raise":
|
||||
window.raise_()
|
||||
window.activateWindow()
|
||||
elif command != "state":
|
||||
print(f"Unknown command: {command}", flush=True)
|
||||
return
|
||||
report(f"after {command}", number)
|
||||
# Observe settled state; this timer does not alter the window or delay the operation.
|
||||
QTimer.singleShot(250, window, lambda: report(f"250ms after {command}", number))
|
||||
|
||||
notifier.activated.connect(read_command)
|
||||
return app.exec()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,15 +1,17 @@
|
||||
import argparse
|
||||
from threading import Thread
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from qtpy.QtCore import QCoreApplication, QEvent, QObject, Qt
|
||||
from qtpy.QtCore import QCoreApplication, QEvent, QObject, Qt, QThread
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.applications import companion_app as companion_app_module
|
||||
from bec_widgets.applications.companion_app import GUIServer
|
||||
from bec_widgets.utils import rpc_server as rpc_server_module
|
||||
from bec_widgets.utils.bec_connector import BECConnector
|
||||
from bec_widgets.utils.bec_dispatcher import QtThreadSafeCallback
|
||||
from bec_widgets.utils.rpc_server import RegistryNotReadyError, RPCServer, SingleshotRPCRepeat
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
@@ -251,6 +253,39 @@ def test_repeated_wayland_raises_share_pending_remap(rpc_server, dummy_widget, m
|
||||
assert events.hide_events == events.show_events == 1
|
||||
|
||||
|
||||
def test_wayland_raise_from_dispatcher_worker_runs_on_gui_thread(
|
||||
rpc_server, dummy_widget, monkeypatch, qtbot, qapp
|
||||
):
|
||||
monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
|
||||
monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
|
||||
dummy_widget.show()
|
||||
events = WindowEvents(dummy_widget)
|
||||
activation_threads = []
|
||||
callback = QtThreadSafeCallback(rpc_server.on_rpc_update)
|
||||
instruction = {
|
||||
"action": "raise",
|
||||
"parameter": {"gui_id": dummy_widget.gui_id, "args": [], "kwargs": {}},
|
||||
}
|
||||
with (
|
||||
patch.object(rpc_server, "send_response") as response,
|
||||
patch.object(
|
||||
dummy_widget,
|
||||
"activateWindow",
|
||||
side_effect=lambda: activation_threads.append(QThread.currentThread()),
|
||||
),
|
||||
):
|
||||
worker = Thread(target=callback, args=(instruction, {"request_id": "raise-from-worker"}))
|
||||
worker.start()
|
||||
worker.join(timeout=2)
|
||||
assert not worker.is_alive()
|
||||
qtbot.waitUntil(lambda: bool(activation_threads) and response.called)
|
||||
|
||||
assert activation_threads == [qapp.thread()]
|
||||
assert events.hide_events == events.show_events == 1
|
||||
assert dummy_widget.isVisible()
|
||||
response.assert_called_once_with("raise-from-worker", True, {"result": None})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["hide", "close", "show"])
|
||||
def test_visibility_request_cancels_pending_wayland_raise(
|
||||
rpc_server, dummy_widget, monkeypatch, qapp, method
|
||||
|
||||
Reference in new issue
Block a user