wip fix e2e test on CI

Author: wyzula-jan

bec_widgets/cli/client.py

@@ -742,6 +742,15 @@ def clear_filters(self) -> "None":
         None
         """
 
+    @rpc_call
+    def state_summary(self) -> "dict[str, dict[str, str]]":
+        """
+        Return the displayed beamline states with their current status and label.
+
+        Returns:
+            dict: Mapping of state name to a dictionary with ``status`` and ``label`` keys.
+        """
+
     @rpc_call
     def remove(self):
         """

bec_widgets/widgets/services/beamline_states/beamline_state_pill.py

@@ -821,6 +821,7 @@ class BeamlineStateManager(BECWidget, QWidget):
         "idle_card_background",
         "set_idle_card_background",
         "clear_filters",
+        "state_summary",
         "remove",
         "attach",
         "detach",
@@ -999,6 +1000,18 @@ def clear_filters(self) -> None:
         self._hidden_expanded = False
         self._apply_filters()
 
+    def state_summary(self) -> dict[str, dict[str, str]]:
+        """
+        Return the displayed beamline states with their current status and label.
+
+        Returns:
+            dict: Mapping of state name to a dictionary with ``status`` and ``label`` keys.
+        """
+        return {
+            name: {"status": pill._status, "label": pill._label}
+            for name, pill in self._state_pills.items()
+        }
+
     @SafeSlot()
     def refresh_states(self) -> None:
         """Fetch the latest cached available beamline states and update the list immediately."""

tests/end-2-end/test_beamline_state_manager_e2e.py

@@ -5,8 +5,8 @@
 import pytest
 from bec_lib.bl_states import DeviceWithinLimitsStateConfig
 
-from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStateManager
-
+# pylint: disable=unused-argument
+# pylint: disable=redefined-outer-name
 # pylint: disable=protected-access
 
 
@@ -16,57 +16,57 @@ def _delete_state_if_present(bec, state_name: str) -> None:
 
 
 @pytest.mark.timeout(100)
-def test_beamline_state_manager_adds_updates_and_deletes_state_e2e(qtbot, bec_client_lib):
+def test_beamline_state_manager_adds_updates_and_deletes_state_e2e(
+    qtbot, bec_client_lib, connected_client_gui_obj
+):
     """
-    Verify the real BEC beamline-state flow is reflected by BeamlineStateManager.
-
-    This test requires the e2e BEC servers and is intended to be run with
-    ``--start-servers``.
+    Verify the real BEC beamline-state flow is reflected by a BeamlineStateManager
+    running in the GUI server, accessed through the dock area RPC interface.
     """
+    gui = connected_client_gui_obj
+    dock_area = gui.bec
     bec = bec_client_lib
     dev = bec.device_manager.devices
     scans = bec.scans
+
     state_name = f"samx_widget_limits_{uuid.uuid4().hex[:8]}"
     config = DeviceWithinLimitsStateConfig(
         name=state_name, device="samx", signal="samx", low_limit=0.0, high_limit=10.0, tolerance=1.0
     )
-    manager = BeamlineStateManager(client=bec)
-    qtbot.addWidget(manager)
-    manager.show()
-    qtbot.waitExposed(manager)
+
+    manager = dock_area.new("BeamlineStateManager")
+    qtbot.waitUntil(lambda: manager._gui_id in gui._server_registry, timeout=5000)
+
+    def state_entry() -> dict[str, str]:
+        return manager.state_summary().get(state_name, {})
 
     _delete_state_if_present(bec, state_name)
 
     try:
         bec.beamline_states.add(config)
 
         qtbot.waitUntil(lambda: hasattr(bec.beamline_states, state_name), timeout=10000)
-        qtbot.waitUntil(lambda: state_name in manager._state_pills, timeout=10000)
-
-        pill = manager._state_pills[state_name]
-        assert pill.state_name == state_name
+        qtbot.waitUntil(lambda: state_name in manager.state_summary(), timeout=10000)
 
         scans.umv(dev.samx, 5, relative=False).wait()
         qtbot.waitUntil(
             lambda: getattr(bec.beamline_states, state_name).get()["status"] == "valid",
             timeout=10000,
         )
-        qtbot.waitUntil(lambda: pill._status == "valid", timeout=10000)
-        assert pill._status_label.text() == "VALID"
-        assert pill._detail_label.text() == "Device samx within limits"
+        qtbot.waitUntil(lambda: state_entry().get("status") == "valid", timeout=10000)
+        assert state_entry()["label"] == "Device samx within limits"
 
         scans.umv(dev.samx, 20, relative=False).wait()
         qtbot.waitUntil(
             lambda: getattr(bec.beamline_states, state_name).get()["status"] == "invalid",
             timeout=10000,
         )
-        qtbot.waitUntil(lambda: pill._status == "invalid", timeout=10000)
-        assert pill._status_label.text() == "INVALID"
-        assert pill._detail_label.text() == "Device samx out of limits"
+        qtbot.waitUntil(lambda: state_entry().get("status") == "invalid", timeout=10000)
+        assert state_entry()["label"] == "Device samx out of limits"
 
         bec.beamline_states.delete(state_name)
         qtbot.waitUntil(lambda: not hasattr(bec.beamline_states, state_name), timeout=10000)
-        qtbot.waitUntil(lambda: state_name not in manager._state_pills, timeout=10000)
+        qtbot.waitUntil(lambda: state_name not in manager.state_summary(), timeout=10000)
 
     finally:
         _delete_state_if_present(bec, state_name)

tests/unit_tests/test_beamline_state_pill.py

@@ -220,6 +220,13 @@ def test_beamline_state_manager_adds_and_removes_pills(qtbot, mocked_client):
     assert beamline_state_manager._state_pills["shutter_open"]._name_label.text() == "Shutter"
     assert not beamline_state_manager._empty_label.isVisible()
 
+    beamline_state_manager._state_pills["limits"].update_state(
+        {"name": "limits", "status": "valid", "label": "Within limits."}, {}
+    )
+    summary = beamline_state_manager.state_summary()
+    assert summary["limits"] == {"status": "valid", "label": "Within limits."}
+    assert summary["shutter_open"]["status"] == "unknown"
+
     beamline_state_manager.update_available_states(
         {
             "states": [