269 lines
8.1 KiB
Python
269 lines
8.1 KiB
Python
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from textwrap import dedent
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from PyQt6.QtGui import QResizeEvent
|
|
from PyQt6.QtWidgets import (
|
|
QApplication,
|
|
QWidget,
|
|
)
|
|
from pytestqt.qtbot import QtBot
|
|
from pytestqt.wait_signal import SignalBlocker
|
|
import pytest
|
|
import tractor
|
|
import trio
|
|
|
|
from piker.service import get_runtime_vars
|
|
from piker.ui import _window
|
|
from piker.ui._exec import (
|
|
QtractorSession,
|
|
start_qtractor,
|
|
)
|
|
from piker.ui._window import MainWindow
|
|
|
|
|
|
class QtractorTestWidget(QWidget):
|
|
'''
|
|
Supply the central-widget resize hook expected by `MainWindow`.
|
|
|
|
'''
|
|
window: MainWindow
|
|
|
|
def on_win_resize(self, event: QResizeEvent) -> None:
|
|
'''
|
|
Accept production `MainWindow.resizeEvent()` notifications.
|
|
|
|
'''
|
|
|
|
|
|
async def run_until_window_close(
|
|
main_widget: QtractorTestWidget,
|
|
exited: list[QtractorTestWidget],
|
|
reg_addr: tuple[str, int],
|
|
runtime_overrides: dict[str, Any],
|
|
|
|
) -> None:
|
|
'''
|
|
Publish readiness and retain the guest task until cancellation.
|
|
|
|
'''
|
|
session: QtractorSession|None = (
|
|
main_widget.window.qtractor_session
|
|
)
|
|
if session is None:
|
|
raise RuntimeError(
|
|
'No Qtractor session owns the test window'
|
|
)
|
|
actor: tractor.Actor = tractor.current_actor()
|
|
assert reg_addr in actor.accept_addrs
|
|
assert tractor.runtime._state._runtime_vars[
|
|
'piker_vars'
|
|
] == runtime_overrides
|
|
session.set_ready()
|
|
try:
|
|
await trio.sleep_forever()
|
|
finally:
|
|
exited.append(main_widget)
|
|
|
|
|
|
def test_qtractor_restarts_after_structured_window_close(
|
|
isolated_ui_state: SimpleNamespace,
|
|
qapp: QApplication,
|
|
qtbot: QtBot,
|
|
reg_addr: tuple[str, int],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
|
|
) -> None:
|
|
'''
|
|
Repeated Qt sessions must not signal or strand the Trio runtime.
|
|
|
|
`MainWindow.closeEvent()` previously sent process-wide SIGINT
|
|
while `run_qtractor()` hid the guest outcome behind blocking
|
|
`app.exec_()`. A caller-owned pytest-qt application therefore
|
|
could not observe a clean startup or close one session without
|
|
interrupting pytest.
|
|
|
|
Start two real Tractor root runtimes in sequence on the
|
|
test-owned registry. Each guest publishes readiness only after
|
|
entering its task, then a real window-close event requests
|
|
cancellation through `QtractorSession`. Completion, the guest
|
|
`finally` block, the empty runtime slot, and successful second
|
|
startup prove structured teardown completed without quitting the
|
|
shared Qt application.
|
|
|
|
Before each start, seed `get_runtime_vars()` with a stale
|
|
loopback registry and conflicting runtime overrides. Previously
|
|
these globals overwrote explicit caller kwargs. Assertions inside
|
|
the guest verify the actual bound address and runtime state before
|
|
readiness, rather than merely checking forwarded arguments.
|
|
|
|
'''
|
|
assert isolated_ui_state.qapp is qapp
|
|
exited: list[QtractorTestWidget] = []
|
|
qt_state: tuple[bool, int, str] = (
|
|
qapp.quitOnLastWindowClosed(),
|
|
qapp.cursorFlashTime(),
|
|
qapp.styleSheet(),
|
|
)
|
|
tractor_kwargs: dict[str, Any] = {
|
|
'registry_addrs': [reg_addr],
|
|
'tractor_runtime_overrides': {
|
|
'piker_test_dir': str(
|
|
isolated_ui_state.test_config_dir
|
|
),
|
|
},
|
|
}
|
|
|
|
_: int
|
|
for _ in range(2):
|
|
monkeypatch.setitem(
|
|
get_runtime_vars(),
|
|
'registry_addrs',
|
|
[('127.0.0.1', 0)],
|
|
)
|
|
monkeypatch.setitem(
|
|
get_runtime_vars(),
|
|
'tractor_runtime_overrides',
|
|
{'stale': True},
|
|
)
|
|
session: QtractorSession = start_qtractor(
|
|
func=run_until_window_close,
|
|
args=(
|
|
exited,
|
|
reg_addr,
|
|
tractor_kwargs['tractor_runtime_overrides'],
|
|
),
|
|
main_widget_type=QtractorTestWidget,
|
|
tractor_kwargs=tractor_kwargs,
|
|
app=qapp,
|
|
)
|
|
qtbot.addWidget(session.window)
|
|
|
|
assert session.app is qapp
|
|
assert not session.owns_app_exec
|
|
assert session.window.godwidget is session.main_widget
|
|
assert session.window.isVisible()
|
|
assert session.outcome is None
|
|
|
|
ready: SignalBlocker = qtbot.waitSignal(
|
|
session.application_ready,
|
|
timeout=30_000,
|
|
)
|
|
ready.wait()
|
|
assert session.is_ready
|
|
assert session.trio_token is not None
|
|
assert session.cancel_scope is not None
|
|
|
|
completed: SignalBlocker
|
|
with qtbot.waitSignal(
|
|
session.completed,
|
|
timeout=30_000,
|
|
) as completed:
|
|
session.window.close()
|
|
|
|
assert completed.args == [session.outcome]
|
|
assert session.outcome is not None
|
|
assert session.outcome.unwrap() is None
|
|
assert session.cancel_scope.cancel_called
|
|
assert exited[-1] is session.main_widget
|
|
assert session.window.qtractor_session is None
|
|
assert _window._qt_win is None
|
|
assert tractor.current_actor(
|
|
err_on_no_runtime=False,
|
|
) is None
|
|
assert not qapp.closingDown()
|
|
assert (
|
|
qapp.quitOnLastWindowClosed(),
|
|
qapp.cursorFlashTime(),
|
|
qapp.styleSheet(),
|
|
) == qt_state
|
|
|
|
|
|
@pytest.mark.parametrize('fail', [False, True])
|
|
def test_run_qtractor_blocking_adapter(
|
|
reg_addr: tuple[str, int],
|
|
tmp_path: Path,
|
|
fail: bool,
|
|
|
|
) -> None:
|
|
'''
|
|
The blocking CLI adapter must drive PyQt6 and unwrap its guest.
|
|
|
|
`run_qtractor()` called the removed `QApplication.exec_()` API,
|
|
failing before Trio could start. Run the real adapter in a fresh
|
|
process with an offscreen application and test-owned registry and
|
|
config. The guest either returns normally or raises a sentinel
|
|
error after registering its window for inspection.
|
|
|
|
Guest completion must quit `QApplication.exec()` without a timer
|
|
or external shutdown. A bounded subprocess wait detects hangs;
|
|
exit status, propagated error identity, and the cleared actor and
|
|
window slots prove both outcome paths finish structured teardown.
|
|
Process isolation keeps the CLI's Qt ownership out of pytest-qt.
|
|
|
|
'''
|
|
script: str = dedent(f'''
|
|
from PyQt6.QtGui import QResizeEvent
|
|
from PyQt6.QtWidgets import QWidget
|
|
import tractor
|
|
from piker.ui import _window
|
|
from piker.ui._exec import run_qtractor
|
|
|
|
class Widget(QWidget):
|
|
def on_win_resize(self, event: QResizeEvent) -> None:
|
|
pass
|
|
|
|
windows: list[QWidget] = []
|
|
error = RuntimeError('qtractor guest sentinel')
|
|
|
|
async def main(widget: Widget) -> None:
|
|
windows.append(widget.window)
|
|
assert {reg_addr!r} in (
|
|
tractor.current_actor().accept_addrs
|
|
)
|
|
if {fail!r}:
|
|
raise error
|
|
|
|
try:
|
|
run_qtractor(
|
|
func=main,
|
|
args=(),
|
|
main_widget_type=Widget,
|
|
tractor_kwargs={{
|
|
'registry_addrs': [{reg_addr!r}],
|
|
'tractor_runtime_overrides': {{
|
|
'piker_test_dir': {str(tmp_path)!r},
|
|
}},
|
|
}},
|
|
)
|
|
except RuntimeError as exc:
|
|
assert {fail!r}
|
|
assert exc is error
|
|
else:
|
|
assert not {fail!r}
|
|
|
|
assert len(windows) == 1
|
|
assert windows[0].qtractor_session is None
|
|
assert _window._qt_win is None
|
|
assert tractor.current_actor(
|
|
err_on_no_runtime=False,
|
|
) is None
|
|
''')
|
|
result: subprocess.CompletedProcess[str] = subprocess.run(
|
|
[sys.executable, '-c', script],
|
|
env={
|
|
**os.environ,
|
|
'QT_QPA_PLATFORM': 'offscreen',
|
|
'XDG_CONFIG_HOME': str(tmp_path / 'config'),
|
|
'XDG_CONFIG_DIRS': str(tmp_path / 'system-config'),
|
|
},
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|