add unit test for fastdeploy/inputs/

xilzy · xilzy · commit 1f8fde26b16e · 2025-10-22T16:04:26.000+08:00
diff --git a/tests/input/test_ernie_processor.py b/tests/input/test_ernie_processor.py
@@ -74,6 +74,70 @@ def test_process_request_dict(self):
         result = self.processor.process_request_dict(request_dict, 100)
         self.assertEqual(result["prompt_token_ids"], [1])
 
+    def test_process_response_dict(self):
+        """测试 process_response_dict 根据 stream 参数调用正确的子方法"""
+        response_dict = {"finished": True, "request_id": "req2", "outputs": {"token_ids": [4, 5]}}
+
+        # 模拟两个子方法
+        self.processor.process_response_dict_streaming = MagicMock(return_value={"result": "stream"})
+        self.processor.process_response_dict_normal = MagicMock(return_value={"result": "normal"})
+
+        # 情况1：stream=True
+        result_stream = self.processor.process_response_dict(response_dict, stream=True)
+        self.processor.process_response_dict_streaming.assert_called_once_with(response_dict)
+        self.assertEqual(result_stream["result"], "stream")
+
+        # 情况2：stream=False
+        result_normal = self.processor.process_response_dict(response_dict, stream=False)
+        self.processor.process_response_dict_normal.assert_called_once_with(response_dict)
+        self.assertEqual(result_normal["result"], "normal")
+
+    def test_process_response(self):
+        """测试 process_response 对完整响应的处理逻辑"""
+        # 构造 mock response_dict 对象
+        mock_outputs = MagicMock()
+        mock_outputs.token_ids = [10, 20, self.processor.tokenizer.eos_token_id]
+        mock_outputs.index = 2
+        mock_response_dict = MagicMock()
+        mock_response_dict.request_id = "req3"
+        mock_response_dict.outputs = mock_outputs
+
+        # 模拟 tokenizer.decode
+        self.processor.tokenizer.decode = MagicMock(return_value="decoded_text")
+
+        # 模拟 reasoning_parser
+        mock_reasoning_parser = MagicMock()
+        mock_reasoning_parser.extract_reasoning_content.return_value = ("reasoning_content", "pure_text")
+        self.processor.reasoning_parser = mock_reasoning_parser
+
+        # 模拟 tool_parser
+        mock_tool_parser = MagicMock()
+        mock_tool_parser.extract_tool_calls.return_value = MagicMock(
+            tools_called=False, tool_calls=None, content="tool_text"
+        )
+        self.processor.tool_parser_obj = MagicMock(return_value=mock_tool_parser)
+
+        # 调用方法
+        result = self.processor.process_response(mock_response_dict)
+
+        # 验证 tokenizer.decode 被正确调用（去掉 eos_token）
+        self.processor.tokenizer.decode.assert_called_once_with([10, 20])
+
+        # 验证 reasoning_parser 被调用并正确赋值
+        mock_reasoning_parser.extract_reasoning_content.assert_called_once()
+        self.assertEqual(result.outputs.text, "pure_text")
+        self.assertEqual(result.outputs.reasoning_content, "reasoning_content")
+
+        # 验证 usage 被正确赋值
+        self.assertIn("completion_tokens", result.usage)
+        self.assertEqual(result.usage["completion_tokens"], 3)
+
+        # 验证 tool_parser 被正确调用
+        mock_tool_parser.extract_tool_calls.assert_called_once()
+
+        # 验证返回结果不为 None
+        self.assertIsNotNone(result)
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/input/test_ernie_vl_processor.py b/tests/input/test_ernie_vl_processor.py
@@ -0,0 +1,143 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from fastdeploy.engine.request import Request
+
+# 导入被测类
+from fastdeploy.input.ernie4_5_vl_processor import Ernie4_5_VLProcessor
+
+
+class TestErnie4_5_VLProcessor(unittest.TestCase):
+    """测试 Ernie4_5_VLProcessor 的主要功能"""
+
+    def setUp(self):
+        """初始化一个带有 mock 依赖的 Processor"""
+        # patch DataProcessor，防止真实加载 tokenizer 或模型
+        dp_patcher = patch("fastdeploy.input.ernie4_5_vl_processor.DataProcessor")
+        self.addCleanup(dp_patcher.stop)
+        MockDP = dp_patcher.start()
+
+        # 模拟 DataProcessor 行为
+        self.mock_dp = MockDP.return_value
+        self.mock_dp.eval.return_value = None
+        self.mock_dp.text2ids.return_value = {
+            "input_ids": [1, 2, 3],
+            "token_type_ids": [0, 0, 0],
+            "position_ids": [[0, 0, 0]],
+            "images": [],
+            "grid_thw": [],
+            "image_type_ids": [],
+            "cur_position": 3,
+        }
+        self.mock_dp.request2ids.return_value = self.mock_dp.text2ids.return_value
+        self.mock_dp.image_patch_id = 999
+        self.mock_dp.spatial_conv_size = 64
+        self.mock_dp.tokenizer = MagicMock()
+        self.mock_dp.tokenizer.pad_token_id = 0
+        self.mock_dp.tokenizer.eos_token_id = 2
+
+        # patch GenerationConfig
+        gen_patcher = patch("fastdeploy.input.ernie4_5_vl_processor.GenerationConfig.from_pretrained")
+        self.addCleanup(gen_patcher.stop)
+        gen_patcher.start()
+
+        # patch Request.from_dict 避免真实依赖
+        req_patcher = patch("fastdeploy.input.ernie4_5_vl_processor.Request.from_dict")
+        self.addCleanup(req_patcher.stop)
+        self.mock_from_dict = req_patcher.start()
+        self.mock_from_dict.side_effect = lambda d: Request(d)
+
+        # 创建 Processor 实例
+        self.processor = Ernie4_5_VLProcessor(model_name_or_path="mock_path")
+
+        # mock 父类 tokenizer
+        self.processor.tokenizer = MagicMock()
+        self.processor.tokenizer.eos_token_id = 2
+        self.processor.tokenizer.pad_token_id = 0
+        self.processor.tokenizer.decode = MagicMock(return_value="decoded text")
+
+    # ----------------------------- #
+    # 测试 process_request_dict
+    # ----------------------------- #
+    def test_process_request_dict_with_prompt(self):
+        """测试含 prompt 的请求"""
+        req = {"prompt": "hello world"}
+        result = self.processor.process_request_dict(req, max_model_len=10)
+
+        self.assertIsInstance(result, dict)
+        self.assertIn("prompt_token_ids", result)
+        self.assertIsInstance(result["prompt_token_ids"], list)
+        self.assertIn("multimodal_inputs", result)
+        self.assertIsInstance(result["multimodal_inputs"], dict)
+        self.assertEqual(result["prompt_token_ids_len"], len(result["prompt_token_ids"]))
+
+    def test_process_request_dict_with_messages(self):
+        """测试含 messages 的请求"""
+        req = {"messages": [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]}
+        result = self.processor.process_request_dict(req)
+        self.assertIn("prompt_token_ids", result)
+        self.assertIn("multimodal_inputs", result)
+
+    # ----------------------------- #
+    # 测试 process_request
+    # ----------------------------- #
+    def test_process_request(self):
+        """测试 process_request 能调用 process_request_dict 并返回正确的 Request"""
+        # 模拟 Request 对象
+        mock_request = MagicMock()
+        mock_request.to_dict.return_value = {"prompt": "test prompt"}
+        self.processor.process_request_dict = MagicMock(
+            return_value={"prompt": "test prompt", "prompt_token_ids": [1, 2]}
+        )
+        self.processor._apply_default_parameters = MagicMock(
+            return_value=Request({"prompt": "test prompt", "prompt_token_ids": [1, 2]})
+        )
+
+        result = self.processor.process_request(mock_request, max_model_len=10)
+        self.processor.process_request_dict.assert_called_once()
+        self.processor._apply_default_parameters.assert_called_once()
+        self.assertIsInstance(result, Request)
+        self.assertEqual(result.data["prompt_token_ids"], [1, 2])
+
+    # ----------------------------- #
+    # 测试 process_response
+    # ----------------------------- #
+    def test_process_response(self):
+        """测试继承自父类的 process_response"""
+        response_dict = MagicMock()
+        response_dict.request_id = "123"
+        response_dict.outputs = MagicMock()
+        response_dict.outputs.token_ids = [1, 2, 3]
+        response_dict.outputs.index = 2
+
+        result = self.processor.process_response(response_dict)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.outputs.text, "decoded text")
+        self.processor.tokenizer.decode.assert_called_once()
+
+    # ----------------------------- #
+    # 测试 process_response_dict
+    # ----------------------------- #
+    def test_process_response_dict_non_stream(self):
+        """测试非流式返回"""
+        mock_normal = MagicMock(return_value={"text": "done"})
+        self.processor.process_response_dict_normal = mock_normal
+
+        response = {"outputs": {"token_ids": [1, 2, 3]}, "finished": True, "request_id": "req_1"}
+        result = self.processor.process_response_dict(response, stream=False)
+        mock_normal.assert_called_once()
+        self.assertEqual(result, {"text": "done"})
+
+    def test_process_response_dict_stream(self):
+        """测试流式返回"""
+        mock_stream = MagicMock(return_value={"delta": "ok"})
+        self.processor.process_response_dict_streaming = mock_stream
+
+        response = {"outputs": {"token_ids": [1, 2, 3]}, "finished": True, "request_id": "req_2"}
+        result = self.processor.process_response_dict(response, stream=True)
+        mock_stream.assert_called_once()
+        self.assertEqual(result, {"delta": "ok"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/input/test_qwen_vl_processor.py b/tests/input/test_qwen_vl_processor.py
@@ -207,6 +207,62 @@ def test_process_request_dict(self):
         self.assertEqual(result["multimodal_inputs"]["pic_cnt"], 1)
         self.assertEqual(result["multimodal_inputs"]["video_cnt"], 1)
 
+    def test_process_response_dict(self):
+        """
+        Test processing of a response dictionary through the processor.
+
+        Ensures:
+        1. The returned dict contains expected keys
+        2. Multimodal outputs are preserved
+        3. Text output matches expected decoded tokens
+        """
+        # Mock output from model
+        response_dict = {
+            "request_id": "12345",
+            "responses": [{"text": "This is a test response."}],
+            "multimodal_outputs": {"images": np.random.randint(0, 256, (1, 3, 224, 224))},
+        }
+
+        # Patch processor's ids2tokens to return expected text
+        with patch.object(
+            self.processor.processor, "ids2tokens", return_value=["This", "is", "a", "test", "response", "."]
+        ):
+            processed = self.processor.process_response_dict(response_dict)
+
+        # Assertions
+        self.assertIn("request_id", processed)
+        self.assertIn("responses", processed)
+        self.assertIn("multimodal_outputs", processed)
+        self.assertEqual(processed["responses"][0]["text"], "This is a test response.")
+
+    def test_process_response(self):
+        """
+        Test processing of a Response object through the processor.
+
+        Ensures:
+        1. Returns a Request object
+        2. Response text is correctly decoded
+        3. Multimodal outputs are preserved
+        """
+        from fastdeploy.engine.response import Response
+
+        # Mock a Response object
+        response = Response(
+            request_id="12345",
+            responses=[{"text": "Another test response"}],
+            multimodal_outputs={"images": np.random.randint(0, 256, (1, 3, 224, 224))},
+        )
+
+        # Patch ids2tokens
+        with patch.object(self.processor.processor, "ids2tokens", return_value=["Another", "test", "response"]):
+            processed = self.processor.process_response(response)
+
+        # Assertions
+        self.assertIsInstance(processed, Request)
+        self.assertEqual(processed.responses[0]["text"], "Another test response")
+        self.assertTrue("multimodal_outputs" in processed.__dict__)
+        self.assertEqual(processed.multimodal_outputs["images"].shape, response.multimodal_outputs["images"].shape)
+
     def test_prompt(self):
         """
         Test processing of prompt with image and video placeholders
diff --git a/tests/input/test_text_processor.py b/tests/input/test_text_processor.py
@@ -58,6 +58,44 @@ def test_process_request_dict(self):
         result = self.processor.process_request_dict(request_dict, 100)
         self.assertEqual(result["prompt_token_ids"], [1])
 
+    def test_process_response_dict(self):
+        # ===== 测试 streaming 分支 =====
+        response_stream = {
+            "request_id": "req_stream",
+            "outputs": {"token_ids": [5, 6, 7]},
+            "finished": False,
+        }
+        # mock ids2tokens 行为
+        self.processor.ids2tokens = MagicMock(return_value=("delta", [5, 6], "prev"))
+        # 确保 streaming 调用
+        result_stream = self.processor.process_response_dict(response_stream, stream=True)
+        self.assertIn("outputs", result_stream)
+        self.assertEqual(result_stream["outputs"]["raw_prediction"], "delta")
+
+        # ===== 测试 normal 分支 =====
+        response_normal = {
+            "request_id": "req_normal",
+            "outputs": {"token_ids": [8, 9, 1]},  # 含 eos_token_id
+            "finished": True,
+        }
+        # mock ids2tokens 行为
+        self.processor.ids2tokens = MagicMock(return_value=("delta", [8, 9], "prev"))
+        self.processor.decode_status["req_normal"] = [0, 0, [], ""]
+        result_normal = self.processor.process_response_dict(response_normal, stream=False)
+        self.assertIn("text", result_normal["outputs"])
+        self.assertEqual(result_normal["outputs"]["text"], "prevdelta")
+
+    def test_process_response(self):
+        # 模拟 response_dict 结构
+        response_mock = MagicMock()
+        response_mock.request_id = "req1"
+        response_mock.outputs = MagicMock()
+        response_mock.outputs.token_ids = [2, 3, 1]  # 含有 eos_token_id
+        # decode 应该去掉 eos_token_id 并返回 "decoded text"
+        result = self.processor.process_response(response_mock)
+        self.processor.tokenizer.decode.assert_called_with([2, 3])
+        self.assertEqual(result.outputs.text, "decoded text")
+
 
 if __name__ == "__main__":
     unittest.main()
