[CI] 【Hackathon 10th Spring No.30】功能模块 fastdeploy/inter_communicator/engine_worker_queue.py单测补充 (#6102)

* test: add comprehensive tests for EngineWorkerQueue to improve code coverage

* style: format tests/inter_communicator/test_e2w_queue.py with black

---------

Co-authored-by: CSWYF3634076 <wangyafeng@baidu.com>

Author: kesmeey

tests/inter_communicator/test_e2w_queue.py

@@ -14,13 +14,20 @@
 # limitations under the License.
 """
 
+import threading
+import time
+import types
 import unittest
 
 import numpy as np
 import paddle
 
+if not hasattr(paddle, "compat"):
+    paddle.compat = types.SimpleNamespace(enable_torch_proxy=lambda **_: None)
+
 from fastdeploy import envs
 from fastdeploy.engine.request import Request
+from fastdeploy.inter_communicator.engine_worker_queue import EngineWorkerQueue
 from fastdeploy.utils import to_numpy, to_tensor
 
 
@@ -30,6 +37,37 @@ def __init__(self, images):
 
 
 class TestEngineWorkerQueue(unittest.TestCase):
+    def _build_queue_pair(self):
+        server = EngineWorkerQueue(address=("127.0.0.1", 0), is_server=True, num_client=1, client_id=0)
+        client = EngineWorkerQueue(
+            address=server.address,
+            is_server=False,
+            num_client=1,
+            client_id=0,
+        )
+        return server, client
+
+    def _cleanup_queue_pair(self, server):
+        server.cleanup()
+
+    def _set_list_after_delay(self, list_proxy, values, delay=0.01):
+        def updater():
+            time.sleep(delay)
+            list_proxy[:] = values
+
+        thread = threading.Thread(target=updater)
+        thread.start()
+        return thread
+
+    def _set_value_after_delay(self, value_proxy, value, delay=0.01):
+        def updater():
+            time.sleep(delay)
+            value_proxy.set(value)
+
+        thread = threading.Thread(target=updater)
+        thread.start()
+        return thread
+
     def test_to_tensor_success(self):
         envs.FD_ENABLE_MAX_PREFILL = 1
         # 模拟 numpy 数组输入（使用 paddle 转 numpy）
@@ -172,6 +210,194 @@ def test_features_info_to_numpy(self):
         self.assertIsInstance(task.multimodal_inputs["video_features"][0], np.ndarray)
         self.assertIsInstance(task.multimodal_inputs["video_features"][1], np.ndarray)
 
+    def test_queue_exist_tasks_and_ports(self):
+        server, client = self._build_queue_pair()
+        try:
+            self.assertIsNone(server.exist_tasks_intra_signal)
+            self.assertFalse(client.exist_tasks())
+            client.set_exist_tasks(True)
+            self.assertTrue(client.exist_tasks())
+            self.assertEqual(server.get_server_port(), server.address[1])
+            with self.assertRaises(RuntimeError):
+                client.get_server_port()
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_single_node_signal_updates(self):
+        server = EngineWorkerQueue(address=("0.0.0.0", 0), is_server=True, num_client=1, client_id=0)
+        try:
+            self.assertFalse(server.exist_tasks())
+            server.set_exist_tasks(True)
+            self.assertTrue(server.exist_tasks())
+            server.set_exist_tasks(False)
+            self.assertFalse(server.exist_tasks())
+        finally:
+            server.cleanup()
+            server.exist_tasks_intra_signal.clear()
+
+    def test_put_get_tasks_and_clear_data(self):
+        envs.FD_ENABLE_MAX_PREFILL = 0
+        envs.FD_ENABLE_E2W_TENSOR_CONVERT = 0
+        server, client = self._build_queue_pair()
+        try:
+            tasks = ["task-A"]
+            client.put_tasks(tasks)
+            self.assertEqual(client.num_tasks(), 1)
+            fetched, all_read = client.get_tasks()
+            self.assertTrue(all_read)
+            self.assertEqual(fetched, [tasks])
+            self.assertEqual(client.num_tasks(), 0)
+            client.put_tasks(tasks)
+            client.clear_data()
+            self.assertEqual(list(client.client_read_flag), [1])
+            self.assertEqual(client.num_tasks(), 0)
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_wait_loops_and_tensor_conversion(self):
+        envs.FD_ENABLE_MAX_PREFILL = 1
+        envs.FD_ENABLE_E2W_TENSOR_CONVERT = 0
+        server, client = self._build_queue_pair()
+        previous_device = paddle.get_device()
+        paddle.set_device("cpu")
+        try:
+            np_images = paddle.randn([1, 3, 4, 4]).numpy()
+            task = DummyTask(np_images)
+            tasks = [[task]]
+            client.client_read_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_read_flag, [1])
+            client.put_tasks(tasks)
+            thread.join()
+            self.assertIsInstance(task.multimodal_inputs["images"], paddle.Tensor)
+
+            client.client_get_connect_task_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_get_connect_task_flag, [1])
+            client.put_connect_rdma_task({"connect": "wait"})
+            thread.join()
+
+            client.can_put_next_connect_task_response_flag.set(0)
+            thread = self._set_value_after_delay(client.can_put_next_connect_task_response_flag, 1)
+            client.put_connect_rdma_task_response({"success": True})
+            thread.join()
+
+            client.connect_rdma_task_responses.append({"success": True})
+            client.client_get_connect_task_response_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_get_connect_task_response_flag, [1])
+            client.get_connect_rdma_task_response()
+            thread.join()
+
+            client.client_read_info_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_read_info_flag, [1])
+            client.put_cache_info([{"cache": "wait"}])
+            thread.join()
+
+            client.can_put_next_send_cache_finished_flag.set(0)
+            thread = self._set_value_after_delay(client.can_put_next_send_cache_finished_flag, 1)
+            client.put_finished_req([["req-wait", {"status": "ok"}]])
+            thread.join()
+
+            client.finished_send_cache_list.append(["req-wait", {"error": "bad"}])
+            client.client_get_finish_send_cache_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_get_finish_send_cache_flag, [1])
+            client.get_finished_req()
+            thread.join()
+
+            client.can_put_next_add_task_finished_flag.set(0)
+            thread = self._set_value_after_delay(client.can_put_next_add_task_finished_flag, 1)
+            client.put_finished_add_cache_task_req(["req-wait"])
+            thread.join()
+
+            client.finished_add_cache_task_list.append(["req-wait"])
+            client.client_get_finished_add_cache_task_flag[:] = [0]
+            thread = self._set_list_after_delay(client.client_get_finished_add_cache_task_flag, [1])
+            client.get_finished_add_cache_task_req()
+            thread.join()
+        finally:
+            paddle.set_device(previous_device)
+            self._cleanup_queue_pair(server)
+
+    def test_connect_rdma_task_flow(self):
+        server, client = self._build_queue_pair()
+        try:
+            client.client_get_connect_task_flag[:] = [1]
+            client.put_connect_rdma_task({"connect": "ok"})
+            task, all_read = client.get_connect_rdma_task()
+            self.assertTrue(all_read)
+            self.assertEqual(task, {"connect": "ok"})
+            self.assertEqual(list(client.connect_rdma_tasks), [])
+
+            self.assertIsNone(client.get_connect_rdma_task_response())
+            response = {"success": True}
+            self.assertTrue(client.put_connect_rdma_task_response(response))
+            client.connect_rdma_task_responses.append({"success": False})
+            merged = client.get_connect_rdma_task_response()
+            self.assertEqual(merged["success"], False)
+            self.assertEqual(client.can_put_next_connect_task_response_flag.get(), 1)
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_cache_info_and_counts(self):
+        server, client = self._build_queue_pair()
+        try:
+            client.client_read_info_flag[:] = [1]
+            cache_info = [{"cache": "info"}]
+            client.put_cache_info(cache_info)
+            self.assertEqual(client.num_cache_infos(), 1)
+            self.assertEqual(client.get_cache_info(), cache_info)
+            self.assertEqual(client.num_cache_infos(), 0)
+            self.assertEqual(client.get_cache_info(), [])
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_finished_req_flow(self):
+        server, client = self._build_queue_pair()
+        try:
+            send_cache_result = [["req-1", {"status": "ok"}]]
+            self.assertTrue(client.put_finished_req(send_cache_result))
+            client.finished_send_cache_list.append(["req-1", {"error": "bad"}])
+            response = client.get_finished_req()
+            self.assertEqual(response, [["req-1", {"error": "bad"}]])
+            self.assertEqual(client.get_finished_req(), [])
+            self.assertEqual(client.can_put_next_send_cache_finished_flag.get(), 1)
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_finished_add_cache_task_req(self):
+        server, client = self._build_queue_pair()
+        try:
+            req_ids = ["req-2"]
+            self.assertTrue(client.put_finished_add_cache_task_req(req_ids))
+            client.finished_add_cache_task_list.append(req_ids)
+            self.assertEqual(client.get_finished_add_cache_task_req(), req_ids)
+            self.assertEqual(client.get_finished_add_cache_task_req(), [])
+            self.assertEqual(client.can_put_next_add_task_finished_flag.get(), 1)
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_disaggregated_queue(self):
+        server, client = self._build_queue_pair()
+        try:
+            self.assertTrue(client.disaggregate_queue_empty())
+            client.put_disaggregated_tasks({"item": 1})
+            client.put_disaggregated_tasks({"item": 2})
+            self.assertFalse(client.disaggregate_queue_empty())
+            self.assertEqual(client.get_disaggregated_tasks(), [{"item": 1}, {"item": 2}])
+            self.assertIsNone(client.get_disaggregated_tasks())
+        finally:
+            self._cleanup_queue_pair(server)
+
+    def test_connect_retry_failure(self):
+        dummy = EngineWorkerQueue.__new__(EngineWorkerQueue)
+
+        class DummyManager:
+            def connect(self):
+                raise ConnectionRefusedError("refused")
+
+        dummy.manager = DummyManager()
+        dummy.address = ("127.0.0.1", 9999)
+        with self.assertRaises(ConnectionError):
+            dummy._connect_with_retry(max_retries=2, interval=0)
+
 
 if __name__ == "__main__":
     unittest.main()