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 tractor import trio 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], ) -> 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' ) 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], ) -> 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. ''' 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): session: QtractorSession = start_qtractor( func=run_until_window_close, args=(exited,), 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