f

Author: wakonig

bec_ipython_client/tests/end-2-end/test_scans_lib_e2e.py

@@ -411,32 +411,40 @@ def test_config_reload_with_describe_failure(bec_test_config_file_path, bec_clie
         },
     }
 
-    # set hexapod to fail
-    bec.connector.set(
-        f"e2e_test_hexapod_fail", messages.DeviceStatusMessage(device="hexapod", status=1)
-    )
+    try:
+        # set hexapod to fail
+        bec.connector.set(
+            "e2e_test_hexapod_fail", messages.DeviceStatusMessage(device="hexapod", status=1)
+        )
 
-    # write new config to disk
-    with open(runtime_config_file_path, "w") as f:
-        f.write(yaml.dump(config))
+        # write new config to disk
+        with open(runtime_config_file_path, "w") as f:
+            f.write(yaml.dump(config))
 
-    with pytest.raises(DeviceConfigError):
-        bec.config.update_session_with_file(runtime_config_file_path, force=True, validate=False)
+        with pytest.raises(DeviceConfigError):
+            bec.config.update_session_with_file(
+                runtime_config_file_path, force=True, validate=False
+            )
 
-    assert len(bec.device_manager.devices) == 2
-    assert bec.device_manager.devices["eyefoc"].enabled is True
-    assert bec.device_manager.devices["hexapod"].enabled is False
+        assert len(bec.device_manager.devices) == 2
+        assert bec.device_manager.devices["eyefoc"].enabled is True
+        assert bec.device_manager.devices["hexapod"].enabled is False
 
-    # set hexapod to pass
-    bec.connector.set(
-        f"e2e_test_hexapod_fail", messages.DeviceStatusMessage(device="hexapod", status=0)
-    )
+        # set hexapod to pass
+        bec.connector.set(
+            "e2e_test_hexapod_fail", messages.DeviceStatusMessage(device="hexapod", status=0)
+        )
 
-    bec.config.update_session_with_file(runtime_config_file_path, force=True)
-    assert len(bec.device_manager.devices) == 2
-    assert bec.device_manager.devices["eyefoc"].enabled is True
-    assert bec.device_manager.devices["hexapod"].enabled is True
-    assert bec.device_manager.devices["hexapod"].precision == 3
+        bec.config.update_session_with_file(runtime_config_file_path, force=True)
+        assert len(bec.device_manager.devices) == 2
+        assert bec.device_manager.devices["eyefoc"].enabled is True
+        assert bec.device_manager.devices["hexapod"].enabled is True
+        assert bec.device_manager.devices["hexapod"].precision == 3
+    finally:
+        bec.connector.set(
+            "e2e_test_hexapod_fail", messages.DeviceStatusMessage(device="hexapod", status=0)
+        )
+        bec.config.update_session_with_file(bec_test_config_file_path, force=True)
 
 
 @pytest.mark.timeout(100)

bec_server/bec_server/scan_server/scan_guard.py

@@ -115,10 +115,10 @@ def _check_valid_scan(self, request: messages.ScanQueueMessage) -> None:
         if scan_type not in avail_scans.resource:
             raise ScanRejection(f"Unknown scan type {scan_type}.")
 
-        if scan_type == "device_rpc":
+        if scan_type in ["device_rpc", "_v4_device_rpc"]:
             # ensure that the requested rpc is allowed for this particular device
-            params = request.content.get("parameter")
-            if not self._device_rpc_is_valid(device=params.get("device"), func=params.get("func")):
+            device, func = self._extract_device_rpc_target(request)
+            if not self._device_rpc_is_valid(device=device, func=func):
                 raise ScanRejection(f"Rejected rpc: {request.content}")
 
     def _device_rpc_is_valid(self, device: str, func: str) -> bool:
@@ -142,10 +142,12 @@ def _check_motors_movable(self, request: messages.ScanQueueMessage) -> None:
         """
         parameter = request.parameter
         if request.scan_type in ["device_rpc", "_v4_device_rpc"]:
-            device = parameter.get("device")
+            device, _ = self._extract_device_rpc_target(request)
             if not isinstance(device, list):
                 device = [device]
             for dev in device:
+                if dev not in self.device_manager.devices:
+                    raise ScanRejection(f"Device {dev} is not known.")
                 if not self.device_manager.devices[dev].enabled:
                     raise ScanRejection(f"Device {dev} is not enabled.")
             return
@@ -212,14 +214,24 @@ def _handle_scan_request(self, msg: messages.ScanQueueMessage, username: str):
             logger.info(f"Request was rejected: {scan_status.message}")
             return
 
-        if msg.scan_type == "device_rpc":
-            func = msg.content.get("parameter", {}).get("func", "")
+        if msg.scan_type in ["device_rpc", "_v4_device_rpc"]:
+            _, func = self._extract_device_rpc_target(msg)
             if func in ["get", "read"] or func.endswith(".get") or func.endswith(".read"):
                 logger.info("Scan request is a read operation, not enqueuing.")
                 self._direct_device_rpc(msg)
                 return
         self._append_to_scan_queue(msg)
 
+    @staticmethod
+    def _extract_device_rpc_target(
+        msg: messages.ScanQueueMessage,
+    ) -> tuple[str | list[str] | None, str]:
+        params = msg.content.get("parameter", {})
+        if msg.scan_type == "_v4_device_rpc":
+            rpc_kwargs = params.get("kwargs", {})
+            return rpc_kwargs.get("device"), rpc_kwargs.get("func", "")
+        return params.get("device"), params.get("func", "")
+
     def _direct_device_rpc(self, msg: messages.ScanQueueMessage):
         """
         Directly send a device RPC request without enqueuing.

bec_server/tests/tests_scan_server/test_scan_guard.py

@@ -42,6 +42,22 @@ def scan_guard_mock(scan_server_mock):
                 queue="primary",
             )
         ),
+        (
+            messages.ScanQueueMessage(
+                scan_type="_v4_device_rpc",
+                parameter={
+                    "args": [],
+                    "kwargs": {
+                        "device": "samy",
+                        "func": "set",
+                        "func_args": [1],
+                        "func_kwargs": {},
+                        "rpc_id": "rpc-id",
+                    },
+                },
+                queue="primary",
+            )
+        ),
     ],
 )
 def test_check_motors_movable_enabled(scan_server_mock, scan_queue_msg):
@@ -97,6 +113,24 @@ def test_device_rpc_is_valid(scan_guard_mock, device, func, is_valid):
             ),
             True,
         ),
+        (
+            messages.ScanQueueMessage(
+                scan_type="_v4_device_rpc",
+                parameter={
+                    "args": [],
+                    "kwargs": {
+                        "device": "samy",
+                        "func": "set",
+                        "func_args": [1],
+                        "func_kwargs": {},
+                        "rpc_id": "rpc-id",
+                    },
+                },
+                queue="primary",
+                metadata={"client_info": {"acl_user": "default"}},
+            ),
+            True,
+        ),
     ],
 )
 def test_valid_request(scan_server_mock, scan_queue_msg, valid):
@@ -316,6 +350,36 @@ def test_handle_scan_request(scan_guard_mock):
             },
             queue="primary",
         ),
+        messages.ScanQueueMessage(
+            metadata={"RID": "9db0c540-c1e0-4f44-872d-f41209512316", "response": True},
+            scan_type="_v4_device_rpc",
+            parameter={
+                "args": [],
+                "kwargs": {
+                    "device": "hexapod",
+                    "rpc_id": "rpc-id-v4",
+                    "func": "x.read",
+                    "func_args": [],
+                    "func_kwargs": {},
+                },
+            },
+            queue="primary",
+        ),
+        messages.ScanQueueMessage(
+            metadata={"RID": "7d28d801-8e33-484d-af34-9cc061eefe0e", "response": True},
+            scan_type="_v4_device_rpc",
+            parameter={
+                "args": [],
+                "kwargs": {
+                    "device": "samx",
+                    "rpc_id": "rpc-id-v4-2",
+                    "func": "get",
+                    "func_args": [],
+                    "func_kwargs": {},
+                },
+            },
+            queue="primary",
+        ),
     ],
 )
 def test_handle_scan_request_bypassed_for_read(scan_guard_mock, msg):