bec_widgets/tests/unit_tests/test_device_browser.py

from typing import TYPE_CHECKING
from unittest import mock

import pytest
from bec_lib.atlas_models import Device as DeviceConfigModel
from bec_lib.device import Device
from qtpy.QtCore import QPoint, Qt
from qtpy.QtWidgets import QDialog, QLabel, QTabWidget

from bec_widgets.applications.views.device_manager_view.device_manager_dialogs import (
    DeviceFormDialog,
)
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.device_browser.device_browser import DeviceBrowser
from bec_widgets.widgets.services.device_browser.device_item.device_item import (
    config_change,
    config_to_yaml,
)
from bec_widgets.widgets.services.device_browser.device_item.device_signal_display import (
    SignalDisplay,
)

if TYPE_CHECKING:  # pragma: no cover
    from qtpy.QtWidgets import QListWidgetItem

    from bec_widgets.widgets.services.device_browser.device_item import DeviceItem


# pylint: disable=no-member
# pylint: disable=missing-function-docstring
# pylint: disable=redefined-outer-name
# pylint: disable=protected-access


@pytest.fixture
def device_browser(qtbot, mocked_client):
    dev_browser = DeviceBrowser(client=mocked_client)
    dev_browser.dev["samx"].read_configuration = mock.MagicMock()
    qtbot.addWidget(dev_browser)
    qtbot.waitExposed(dev_browser)
    yield dev_browser


def test_device_browser_init_with_devices(device_browser: DeviceBrowser):
    """
    Test that the device browser is initialized with the correct number of devices.
    """
    device_list = device_browser.dev_list
    assert device_list.count() == len(device_browser.dev)


@pytest.mark.parametrize(
    ["search_term", "expected_num_visible"],
    [("sam", 3), ("nonexistent", 0), ("", -1), (r"(\)", -1)],
)
def test_device_browser_filtering(
    qtbot, device_browser, search_term: str, expected_num_visible: int
):
    """
    Test that the device browser is able to filter the device list.
    """
    expected = expected_num_visible if expected_num_visible >= 0 else len(device_browser.dev)

    def num_visible(item_dict):
        return len(list(filter(lambda i: not i.widget.isHidden(), item_dict.values())))

    device_browser.ui.filter_input.setText(search_term)
    qtbot.wait(100)
    assert num_visible(device_browser.dev_list._item_dict) == expected


def test_device_item_mouse_press_event(device_browser, qtbot):
    """
    Test that the mousePressEvent is triggered correctly.
    """
    # Simulate a left mouse press event on the device item
    device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
    widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
    qtbot.mouseClick(widget._title, Qt.MouseButton.LeftButton)


def test_update_event_captured(device_browser, qtbot):
    device_browser.update_device_list = mock.MagicMock()
    device_browser.update_device_list.assert_not_called()
    device_browser.on_device_update("remove", {})
    device_browser.update_device_list.assert_called_once()
    device_browser.on_device_update("", {})


def test_device_item_expansion(device_browser, qtbot):
    """
    Test that the form is displayed when the item is expanded, and that the expansion is triggered
    by clicking on the expansion button, the title, or the device icon
    """
    device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
    widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
    qtbot.mouseClick(widget._expansion_button, Qt.MouseButton.LeftButton)
    tab_widget: QTabWidget = widget._contents.layout().itemAt(0).widget()
    qtbot.waitUntil(lambda: tab_widget.widget(0) is not None, timeout=100)
    qtbot.waitUntil(
        lambda: isinstance(tab_widget.widget(0).layout().itemAt(0).widget(), QLabel), timeout=100
    )
    config_view: QLabel = tab_widget.widget(0).layout().itemAt(0).widget()
    assert widget.expanded
    qtbot.waitUntil(lambda: config_view.text().startswith("name: aptrx\n"), timeout=500)
    assert "deviceClass: ophyd_devices.SimPositioner" in config_view.text()
    qtbot.mouseClick(widget._expansion_button, Qt.MouseButton.LeftButton)
    assert not widget.expanded

    qtbot.mouseClick(widget._title, Qt.MouseButton.LeftButton)
    qtbot.waitUntil(lambda: widget.expanded, timeout=500)

    qtbot.mouseClick(widget._title_icon, Qt.MouseButton.LeftButton)
    qtbot.waitUntil(lambda: not widget.expanded, timeout=500)


def test_device_item_mouse_press_and_move_events_creates_drag(device_browser, qtbot):
    """
    Test that the mousePressEvent is triggered correctly and initiates a drag.
    """
    device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
    widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
    device_name = widget.device
    with mock.patch("qtpy.QtGui.QDrag.exec_") as mock_exec:
        with mock.patch("qtpy.QtGui.QDrag.setMimeData") as mock_set_mimedata:
            qtbot.mousePress(widget._title, Qt.MouseButton.LeftButton, pos=QPoint(0, 0))
            qtbot.mouseMove(widget, pos=QPoint(10, 10))
            qtbot.mouseRelease(widget, Qt.MouseButton.LeftButton)
            mock_set_mimedata.assert_called_once()
            mock_exec.assert_called_once()
            assert mock_set_mimedata.call_args[0][0].text() == device_name


def test_device_item_double_click_event(device_browser, qtbot):
    """
    Test that the mouseDoubleClickEvent is triggered correctly.
    """
    # Simulate a left mouse press event on the device item
    device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
    widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
    qtbot.mouseDClick(widget, Qt.LeftButton)


def test_device_deletion(device_browser, qtbot):
    device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
    widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
    widget._config_helper = mock.MagicMock()

    assert widget.device in device_browser.dev_list._item_dict
    qtbot.mouseClick(widget.delete_button, Qt.LeftButton)
    qtbot.waitUntil(lambda: widget.device not in device_browser.dev_list._item_dict, timeout=10000)


def test_signal_display(mocked_client, qtbot):
    signal_display = SignalDisplay(client=mocked_client, device="test_device")
    qtbot.addWidget(signal_display)
    device_mock = mock.MagicMock()
    signal_display.dev = {"test_device": device_mock}
    signal_display._refresh()
    device_mock.read.assert_called()
    device_mock.read_configuration.assert_called()


def test_signal_display_no_device(mocked_client, qtbot):
    device_mock = mock.MagicMock()
    mocked_client.device_manager.devices = {"test_device_1": device_mock}
    signal_display = SignalDisplay(client=mocked_client, device="test_device_2")
    qtbot.addWidget(signal_display)
    assert (
        signal_display._content_layout.itemAt(1).widget().text()
        == "Device test_device_2 not found in device manager!"
    )
    signal_display._refresh()
    device_mock.read.assert_not_called()
    device_mock.read_configuration.assert_not_called()


def test_signal_display_omitted_not_added(mocked_client, qtbot):
    device_mock = mock.MagicMock(spec=Device)
    device_mock._info = {"signals": {"signal_1": {"kind_str": "omitted"}}}

    signal_display = SignalDisplay(client=mocked_client, device="test_device_1")
    signal_display.dev = {"test_device_1": device_mock}
    signal_display._populate()

    qtbot.addWidget(signal_display)
    assert signal_display._content_layout.itemAt(1).widget() is None


def test_device_browser_cleanup_detaches_filter_proxy(device_browser: DeviceBrowser, qtbot):
    """
    A filter change still queued in the rate-limited filter proxy when the browser is
    cleaned up, or made afterwards, must not reach update_device_list of the closed browser.
    """
    proxy = device_browser.proxy_device_update
    with mock.patch.object(device_browser.dev_list, "update_filter") as update_filter:
        device_browser.ui.filter_input.setText("sam")  # queued in the proxy
        assert proxy.timer.timer.isActive()  # pg's ThreadsafeTimer wraps a QTimer
        device_browser.close()  # closeEvent -> cleanup()
        device_browser.ui.filter_input.setText("samx")  # changed after cleanup

        qtbot.wait(600)
        update_filter.assert_not_called()
        assert not proxy.timer.timer.isActive()
        assert proxy.args is None


_BASIC_CONFIG = {
    "name": "test_device",
    "enabled": True,
    "deviceClass": "TestDevice",
    "readoutPriority": "monitored",
    "deviceConfig": {"param": 1},
}


@pytest.mark.parametrize(
    ["edited", "expected_keys"],
    [
        ({}, set()),
        (_BASIC_CONFIG, set()),
        ({"readOnly": False}, set()),
        ({"readOnly": True}, {"readOnly"}),
        ({"readOnly": True, "description": "test"}, {"readOnly", "description"}),
        ({"deviceTags": ["tag"]}, {"deviceTags"}),
        ({"deviceConfig": {"param": 2}}, "all"),
        ({"deviceClass": "OtherDevice"}, "all"),
        ({"name": "renamed"}, "all"),
    ],
)
def test_config_change(edited, expected_keys):
    change = config_change(_BASIC_CONFIG, edited)
    if expected_keys == "all":
        assert change == DeviceConfigModel.model_validate({**_BASIC_CONFIG, **edited}).model_dump()
    else:
        assert set(change) == expected_keys


def test_config_to_yaml_puts_name_and_class_first():
    text = config_to_yaml(DeviceConfigModel.model_validate(_BASIC_CONFIG))
    assert text.splitlines()[:2] == ["name: test_device", "deviceClass: TestDevice"]
    assert "  param: 1" in text


def _opened_dialog(open_dialog) -> DeviceFormDialog:
    opened = []
    with mock.patch.object(QDialog, "open", lambda dialog: opened.append(dialog)):
        open_dialog()
    (dialog,) = opened
    assert isinstance(dialog, DeviceFormDialog)
    return dialog


def _first_item(device_browser) -> "DeviceItem":
    return device_browser.dev_list.itemWidget(device_browser.dev_list.itemAt(0, 0))


def _run_inline(widget):
    widget._q_threadpool = mock.MagicMock()
    widget._q_threadpool.start.side_effect = lambda runnable: runnable.run()
    widget._config_helper = mock.MagicMock()


def test_add_dialog_sends_add_request(device_browser, qtbot):
    dialog = _opened_dialog(device_browser._create_add_dialog)
    _run_inline(device_browser)
    config = {**_BASIC_CONFIG, "name": "new_device"}
    dialog.accepted_data.emit(config, 0, 0, "", "")
    device_browser._config_helper.send_config_request.assert_called_once_with(
        action="add", config={"new_device": config}, wait_for_response=False
    )
    dialog.close()


def test_add_dialog_refuses_existing_name(device_browser, qtbot):
    dialog = _opened_dialog(device_browser._create_add_dialog)
    _run_inline(device_browser)
    with mock.patch(
        "bec_widgets.utils.error_popups._ErrorPopupUtility.custom_exception_hook"
    ) as mock_hook:
        dialog.accepted_data.emit({**_BASIC_CONFIG, "name": "samx"}, 0, 0, "", "")
    assert isinstance(mock_hook.call_args[0][1], ValueError)
    device_browser._config_helper.send_config_request.assert_not_called()
    dialog.close()


def test_edit_dialog_is_prefilled(device_browser, qtbot):
    widget = _first_item(device_browser)
    dialog = _opened_dialog(widget._create_edit_dialog)
    config = dialog._device_config_template.get_config_fields()
    assert config["name"] == widget.device
    assert config["deviceClass"] == device_browser.dev[widget.device]._config["deviceClass"]
    dialog.close()


def test_edit_dialog_sends_only_changes(device_browser, qtbot):
    widget = _first_item(device_browser)
    dialog = _opened_dialog(widget._create_edit_dialog)
    _run_inline(widget)
    config = {**dialog._device_config_template.get_config_fields(), "readOnly": True}
    dialog.accepted_data.emit(config, 0, 0, "", "")
    widget._config_helper.send_config_request.assert_called_once_with(
        action="update", config={widget.device: {"readOnly": True}}, wait_for_response=False
    )
    dialog.close()


def test_edit_dialog_without_changes_sends_nothing(device_browser, qtbot):
    widget = _first_item(device_browser)
    dialog = _opened_dialog(widget._create_edit_dialog)
    _run_inline(widget)
    dialog.accepted_data.emit(dialog._device_config_template.get_config_fields(), 0, 0, "", "")
    widget._config_helper.send_config_request.assert_not_called()
    dialog.close()


def test_edit_dialog_rename_recreates_device(device_browser, qtbot):
    widget = _first_item(device_browser)
    old_name = widget.device
    dialog = _opened_dialog(widget._create_edit_dialog)
    _run_inline(widget)
    config = {**dialog._device_config_template.get_config_fields(), "name": "renamed_device"}
    dialog.accepted_data.emit(config, 0, 0, "", "")
    calls = widget._config_helper.send_config_request.call_args_list
    assert [(c.kwargs["action"], list(c.kwargs["config"])) for c in calls] == [
        ("remove", [old_name]),
        ("add", ["renamed_device"]),
    ]
    qtbot.waitUntil(lambda: old_name not in device_browser.dev_list._item_dict, timeout=1000)
    dialog.close()


def test_edit_dialog_rename_refuses_existing_name(device_browser, qtbot):
    widget = _first_item(device_browser)
    other = next(name for name in device_browser.dev if name != widget.device)
    dialog = _opened_dialog(widget._create_edit_dialog)
    _run_inline(widget)
    config = {**dialog._device_config_template.get_config_fields(), "name": other}
    with mock.patch(
        "bec_widgets.utils.error_popups._ErrorPopupUtility.custom_exception_hook"
    ) as mock_hook:
        dialog.accepted_data.emit(config, 0, 0, "", "")
    assert isinstance(mock_hook.call_args[0][1], ValueError)
    widget._config_helper.send_config_request.assert_not_called()
    dialog.close()