piker/tests/ui/test_qtractor.py

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