AareDAQ/tests/unit/gui/test_main_window.py

from unittest.mock import MagicMock, patch

import pytest
from PySide6.QtCore import QSettings, Qt
from PySide6.QtWidgets import QApplication, QDockWidget

from aare.gui import styles
from aare.gui.main_window import MainWindow
from aare.gui.styles import THEME_BLUEBIRD, THEME_SUNRISE, THEME_SUNSET


@pytest.fixture
def mock_ui_state():
    with patch("aare.gui.main_window.UIStateManager") as mock:
        yield mock


def test_main_window_init(qtbot, mock_ui_state, daq_status_factory):
    with (
        patch("requests.get") as mock_get,
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"status": "ok"}
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }

        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url="http://localhost:5210",
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr="localhost",
            gonio_cam_addr="localhost",
            gonio_cam_id=1,
        )
        win.show()
        qtbot.addWidget(win)

        assert win.windowTitle() == "AareGUI"
        assert win.isVisible()

        # Staff-side gate behaviors piggyback on this window: every extra
        # MainWindow construction raises the odds of the PySide SystemError
        # flake, so don't build another one just for these.
        from aarecommon.models.models import BeamlineStateEnum

        with patch("aare.gui.main_window.QMessageBox") as popup:
            win._show_beamline_tab_gated_popup()
            assert popup.information.called, "staff popup must explain the visitor pgroup"

        win._apply_default_sample_tab(BeamlineStateEnum.BeamLocation)
        assert win.sample_lists_tabs.currentIndex() == 1  # Auxiliary puck
        win._apply_default_sample_tab(BeamlineStateEnum.DataCollection)
        assert win.sample_lists_tabs.currentIndex() == 0

        # Energy row inside Exp. Config. emits eV (the GUI shows keV)
        sent = []
        win.data_collection.change_energy.connect(sent.append)
        win.data_collection.energy_spin.setValue(12.4)
        win.data_collection._emit_change_energy()
        assert sent and abs(sent[0] - 12400.0) < 1e-6

        # Side panels collapse from the View menu (no on-screen buttons), and
        # the camera region sits in a splitter so its width is drag-resizable.
        assert win.center_splitter.widget(1) is win.video_tab
        # The clamp lets the splitter shrink the camera region well below the
        # natural minimum the tab labels + controls rows would demand.
        assert 0 < win.video_tab.minimumWidth() < win.video_tab.minimumSizeHint().width()
        win._show_left_panel_action.trigger()
        assert win.collection_controls_scroll.isHidden()
        win._show_left_panel_action.trigger()
        assert not win.collection_controls_scroll.isHidden()
        win._show_right_panel_action.trigger()
        assert win.beamline_controls_scroll.isHidden()
        win._show_right_panel_action.trigger()
        assert not win.beamline_controls_scroll.isHidden()

        # Alert routers land straight in the runtime notification dock (the
        # portrait-mode interceptors are gone with Playlist Mode).
        with patch.object(win, "_show_runtime_notification") as note:
            win._runtime_alert_primary("beam lost", True)
            win._runtime_alert_secondary("shutter", False)
            win._on_http_error("boom")
            win._on_sample_camera_error("no frames")
            assert note.call_count == 4

        assert (
            win._detector_error_banner_text(automation=True, message="x")
            == "Automation halted: detector error."
        )
        assert (
            win._detector_error_banner_text(automation=False, message="x")
            == "Manual collection stopped: detector error."
        )

        # Manual unmount respects the hutch PSS gate.
        with (
            patch.object(win, "_hutch_blocks_mount", return_value="hutch open"),
            patch("aare.gui.main_window.QMessageBox"),
        ):
            win._on_manual_unmount_requested()
        win.daq.unmount.assert_not_called()
        with patch.object(win, "_hutch_blocks_mount", return_value=None):
            win._on_manual_unmount_requested()
        win.daq.unmount.assert_called_once()

        # Motion watch: only the robot station switches to the combined
        # beamline view. Moving no longer does (users kept losing the sample
        # camera on short gonio moves), and busy alone never does — Sample
        # alignment, Beam location etc. run busy while the user watches the
        # sample camera itself.
        win.update_daq_status(
            daq_status_factory(state=BeamlineStateEnum.RobotSampleExchange, busy=True)
        )
        assert win._watching_motion
        assert win.video_tab.currentWidget() is win.beamline_combined_panel
        win.update_daq_status(
            daq_status_factory(state=BeamlineStateEnum.SampleAlignment, busy=True)
        )
        assert not win._watching_motion
        assert win.video_tab.currentWidget() is win.sample_camera
        win.update_daq_status(daq_status_factory(state=BeamlineStateEnum.BeamLocation, busy=True))
        assert not win._watching_motion
        win.update_daq_status(daq_status_factory(state=BeamlineStateEnum.Moving, busy=True))
        assert not win._watching_motion
        assert win.video_tab.currentWidget() is win.sample_camera
        win.update_daq_status(
            daq_status_factory(state=BeamlineStateEnum.DataCollection, busy=False)
        )
        assert not win._watching_motion
        assert win.video_tab.currentWidget() is win.sample_camera


def test_main_window_mount_view(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url=None,
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr=None,
            gonio_cam_addr=None,
            gonio_cam_id=None,
        )
        qtbot.addWidget(win)

        win.mount_view()


def test_mark_user_interaction_reports_backend(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url=None,
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr=None,
            gonio_cam_addr=None,
            gonio_cam_id=None,
        )
        qtbot.addWidget(win)

        win._last_interaction_report_ts = -999.0
        win._mark_user_interaction()

        win.daq.report_gui_interaction.assert_called_once_with(15)


def test_automation_activity_refreshes_idle_timestamp(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        from aarecommon.math.coordinate import Coordinate, SmargonCoordinate
        from aarecommon.math.diffraction_geometry import DiffractionGeometry
        from aarecommon.math.sample_geometry import SampleGeometryModel
        from aarecommon.models.models import (
            BeamlineStateEnum,
            BeamlineStatus,
            CrystalSize,
            DAQStatusModel,
            SampleCameraSettings,
            SessionsStateEnum,
            SessionStatus,
        )

        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url=None,
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr=None,
            gonio_cam_addr=None,
            gonio_cam_id=None,
        )
        qtbot.addWidget(win)

        win.job_list_panel.is_running = MagicMock(return_value=True)
        win._last_user_interaction_ts = 100.0

        status = DAQStatusModel(
            geom=SampleGeometryModel(
                beam_location_pxl=Coordinate(x=1000, y=1000),
                pixel_in_mm=0.001,
                aerotech=Coordinate(),
                aerotech_meas=Coordinate(),
                smargon=SmargonCoordinate(sh_mm=Coordinate(), phi_deg=0, chi_deg=0),
                omega_deg=0,
                beam_size_mm=Coordinate(x=0.01, y=0.01),
            ),
            diffraction=DiffractionGeometry(
                energy_keV=12.4,
                dtz_mm=100.0,
                detector_size_pxl=(1553, 1630),
                pixel_size_mm=0.150,
                beam_center_pxl=(750.0, 750.0),
                detector_description="PILATUS 4",
                detector_serial_number="1",
                poni_rot1_rad=0.0,
                poni_rot2_rad=0.0,
            ),
            bl=BeamlineStatus(
                name="SIMULATED",
                ring_current_mA=400.0,
                front_light=50.0,
                back_light=50.0,
                cryojet_K=100.0,
                shutter_open=False,
                exp_shutter_open=False,
                flux_ph_s=1e12,
                sample_camera=SampleCameraSettings(gain=1.0, exposure=0.02),
                transmission=1.0,
                zoom=1.0,
                commissioning_mode=False,
                dtz_min=120.0,
                dtz_max=1600.0,
            ),
            state=BeamlineStateEnum.SampleAlignment,
            busy=False,
            session=SessionStatus(
                session=SessionsStateEnum.Vacant, current_pgroup="p123", staff=True
            ),
            crystal_size=CrystalSize(x=0, y=0, z=0),
        )

        with patch("aare.gui.main_window.time.time", return_value=250.0):
            win.update_daq_status(status)

        assert win._last_user_interaction_ts == 250.0

        # Remote-close bookkeeping in the same status loop: an unreadable
        # session entry is skipped, a close request for THIS session starts
        # the countdown, and its disappearance clears it again.
        from types import SimpleNamespace

        unreadable = status.model_copy(update={"open_guis": [SimpleNamespace(session="bad")]})
        win.update_daq_status(unreadable)
        assert win._remote_close_deadline_ts is None

        close_req = status.model_copy(
            update={
                "open_guis": [
                    SimpleNamespace(
                        session=15,
                        close_requested=True,
                        close_requested_by="ops",
                        close_grace_seconds=5,
                    )
                ]
            }
        )
        win.update_daq_status(close_req)
        assert win._remote_close_deadline_ts is not None

        win.update_daq_status(status)
        assert win._remote_close_deadline_ts is None


def test_idle_timeout_closes_when_inactive_and_not_running(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url=None,
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr=None,
            gonio_cam_addr=None,
            gonio_cam_id=None,
        )
        qtbot.addWidget(win)

        win._idle_close_timeout_s = 10
        win._last_user_interaction_ts = 100.0
        win.close = MagicMock()

        with patch("aare.gui.main_window.time.time", return_value=111.0):
            win._check_idle_timeout()

        win.close.assert_called_once()


def test_idle_timeout_does_not_close_while_automation_active(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        fake_token = "header.payload.signature"

        win = MainWindow(
            base_url=None,
            token=fake_token,
            default_image=None,
            zmq_addr=None,
            pred_zmq_addr=None,
            beamline_cam_addr=None,
            gonio_cam_addr=None,
            gonio_cam_id=None,
        )
        qtbot.addWidget(win)

        win._idle_close_timeout_s = 10
        win._last_user_interaction_ts = 100.0
        win.job_list_panel.is_running = MagicMock(return_value=True)
        win.close = MagicMock()

        with patch("aare.gui.main_window.time.time", return_value=111.0):
            win._check_idle_timeout()

        win.close.assert_not_called()


def _make_window(qtbot):
    win = MainWindow(
        base_url=None,
        token="header.payload.signature",
        default_image=None,
        zmq_addr=None,
        pred_zmq_addr=None,
        beamline_cam_addr=None,
        gonio_cam_addr=None,
        gonio_cam_id=None,
    )
    qtbot.addWidget(win)
    return win


def test_theme_settings_migrate_and_slots_switch(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

        settings = QSettings("PSI", "AareGUI")
        saved = settings.value("appearance/theme")
        try:
            # Pre-rename tokens saved by older builds must map to the new ones.
            settings.setValue("appearance/theme", "portrait")
            win._restore_theme_settings()
            assert win._theme_mode == THEME_SUNSET

            settings.setValue("appearance/theme", "original")
            win._restore_theme_settings()
            assert win._theme_mode == THEME_SUNRISE

            settings.setValue("appearance/theme", THEME_BLUEBIRD)
            win._restore_theme_settings()
            assert win._theme_mode == THEME_BLUEBIRD
        finally:
            if saved is None:
                settings.remove("appearance/theme")
            else:
                settings.setValue("appearance/theme", saved)

        win.use_bluebird_theme()
        assert win._theme_mode == THEME_BLUEBIRD
        win.use_portrait_theme()  # exercises the sunset palette flip
        assert win._theme_mode == THEME_SUNSET
        win.use_legacy_theme()
        assert win._theme_mode == THEME_SUNRISE


def test_font_zoom_steps_clamps_and_resets(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

        settings = QSettings("PSI", "AareGUI")
        saved = settings.value("appearance/font_scale")
        try:
            base_pt = win._default_app_font.pointSizeF()
            win._change_font_zoom(1)
            assert styles.font_scale() == pytest.approx(1.25)
            # Part 2 of the zoom: the app default font scales with the ladder.
            app = QApplication.instance()
            assert isinstance(app, QApplication)  # narrow from QCoreApplication|None
            assert app.font().pointSizeF() == pytest.approx(base_pt * 1.25)
            # Part 3: the fixed-width side boxes follow the ladder, else the
            # zoomed text clips inside them.
            assert win.data_collection.minimumWidth() == round(win.data_collection.set_width * 1.25)
            assert win.collection_controls_scroll.maximumWidth() == (
                round(win.data_collection.set_width * 1.25) + 10
            )
            assert win.beamline.minimumWidth() == round(win.beamline.set_width * 1.25)
            for _ in range(5):
                win._change_font_zoom(1)
            assert styles.font_scale() == styles.FONT_SCALE_LADDER[-1]
            win._change_font_zoom(0)
            assert styles.font_scale() == 1.0
            assert app.font().pointSizeF() == pytest.approx(base_pt)
            assert win.data_collection.minimumWidth() == win.data_collection.set_width
            for _ in range(5):
                win._change_font_zoom(-1)
            assert styles.font_scale() == styles.FONT_SCALE_LADDER[0]  # floor is 100%
        finally:
            styles.set_font_scale(1.0)
            if saved is None:
                settings.remove("appearance/font_scale")
            else:
                settings.setValue("appearance/font_scale", saved)


def test_restore_window_state_heals_all_hidden_docks(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

        for dock in win.findChildren(QDockWidget):
            dock.hide()
        assert all(d.isHidden() for d in win.findChildren(QDockWidget))

        # state_manager is mocked, so restore_window is a no-op and the
        # all-hidden layout survives to the heal check.
        win._restore_window_state()

        assert not win.tell_samples_dock.isHidden()


def test_close_restores_pre_watch_layout(qtbot, mock_ui_state):
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

        pre_watch = win.saveState()
        for dock in win.findChildren(QDockWidget):
            dock.hide()
        win._session_operations_enabled = False
        win._pre_watch_dock_state = pre_watch

        win.close()

        # closeEvent put the pre-watch layout back before saving state, so
        # the all-hidden fold was not persisted.
        assert not win.tell_samples_dock.isHidden()


def test_nonstaff_beamline_gate_popups(qtbot, mock_ui_state):
    """Non-staff must SEE the staff-only Beamline panels (as locked banners)
    and get an explanation on click — both on a banner and on the
    pgroup-gated (disabled) Beamline tab, which used to eat clicks silently."""
    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "visitor",
            "staff": False,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

        # Locked stand-ins exist in place of the real staff panels
        titles = [b.text() for b in win._locked_beamline_banners]
        assert titles == ["Beamline setup", "ABR meas. pos.", "Beam configuration"]
        assert not hasattr(win, "monochromator_panel")

        # Staff-gate denials are passive red tips now, not QMessageBoxes.
        with patch("aare.gui.main_window.QToolTip") as tip:
            qtbot.mousePress(win._locked_beamline_banners[0], Qt.MouseButton.LeftButton)
            assert tip.showText.called, "banner click must explain the staff gate"
        # The tip anchor carries the red warning wash (per-widget QToolTip QSS).
        assert "QToolTip" in win._admin_tip_anchor.styleSheet()

        # Active pgroup outside the token disables the whole Beamline tab;
        # a click on the disabled tab must explain itself, not vanish.
        win._apply_pgroup_gate("p999")
        assert not win.left_column_tabs.isTabEnabled(0)
        bar = win.left_column_tabs.tabBar()
        with patch("aare.gui.main_window.QToolTip") as tip:
            qtbot.mousePress(bar, Qt.MouseButton.LeftButton, pos=bar.tabRect(0).center())
            assert tip.showText.called, "gated tab click must explain the gate"

        # The greyed-out Auxiliary-puck tab explains itself the same way.
        aux_bar = win.sample_lists_tabs.tabBar()
        with patch("aare.gui.main_window.QToolTip") as tip:
            qtbot.mousePress(aux_bar, Qt.MouseButton.LeftButton, pos=aux_bar.tabRect(1).center())
            assert tip.showText.called, "aux-puck tab click must explain the lock"


def test_sample_camera_frame_paints_visible_views_and_acks(qtbot, mock_ui_state, monkeypatch):
    from PySide6.QtGui import QImage

    with (
        patch("requests.get"),
        patch("aare.gui.main_window.DAQWorker"),
        patch("aare.gui.main_window.PredictionSubscriber"),
        patch("aare.gui.main_window.VideoThread"),
        patch("aare.gui.main_window.JFJochDBusClient"),
        patch("aare.gui.main_window.jwt.decode") as mock_jwt,
    ):
        mock_jwt.return_value = {
            "sub": "testuser",
            "staff": True,
            "pgroups": ["p123"],
            "session": 15,
        }
        win = _make_window(qtbot)

    # Nothing is shown in tests, so pretend the view is on screen to reach
    # the paint branch.
    monkeypatch.setattr(type(win.sample_camera), "isVisible", lambda self: True)
    win.prediction_thread = MagicMock()

    win._on_sample_camera_frame(QImage(4, 6, QImage.Format.Format_RGB888))

    assert win.sample_camera.pixmap_item.pixmap().width() == 4
    # The ack is what paces the subscriber; it must fire after every frame.
    win.prediction_thread.notify_frame_displayed.assert_called_once()

    # No subscriber (GUI started without a sample feed): the slot must not blow up.
    win.prediction_thread = None
    win._on_sample_camera_frame(QImage(4, 6, QImage.Format.Format_RGB888))