Method test_benchmark

tests/benchmarks/test_serve_benchmarks.py:139–194

(self, mock_gather, mock_get_request)

    @patch("fastdeploy.benchmarks.serve.get_request")
    @patch("asyncio.gather", new_callable=AsyncMock)
    async def test_benchmark(self, mock_gather, mock_get_request):
        # 直接在测试中设置ASYNC_REQUEST_FUNCS
        from fastdeploy.benchmarks.serve import ASYNC_REQUEST_FUNCS

        mock_func = AsyncMock()
        ASYNC_REQUEST_FUNCS["test_backend"] = mock_func
        from fastdeploy.benchmarks.datasets import SampleRequest

        # 创建一个异步生成器函数来模拟get_request
        async def mock_request_gen():
            yield SampleRequest(
                no=1, prompt="test", prompt_len=10, expected_output_len=20, history_QA=[], json_data=None
            )

        mock_get_request.return_value = mock_request_gen()
        mock_func.return_value = MagicMock(
            success=True,
            prompt_len=10,
            prompt_tokens=10,
            output_tokens=20,
            ttft=0.1,
            itl=[0.02, 0.02, 0.02],
            latency=0.5,
            arrival_time=[0, 0.1, 0.12, 0.14, 0.16],
            generated_text="test output",
            reasoning_content=None,
            error=None,
        )

        result = await benchmark(
            backend="test_backend",
            api_url="http://test",
            base_url="http://test",
            model_id="test_model",
            model_name="test_model",
            input_requests=[
                SampleRequest(
                    no=1, prompt="test", prompt_len=10, expected_output_len=20, history_QA=[], json_data=None
                )
            ],
            hyper_parameters={},
            logprobs=None,
            request_rate=1.0,
            burstiness=1.0,
            disable_tqdm=True,
            profile=False,
            selected_percentile_metrics=["ttft", "tpot", "itl"],
            selected_percentiles=[99],
            ignore_eos=False,
            debug=False,
            goodput_config_dict={},
            max_concurrency=None,
            lora_modules=None,
            extra_body=None,
        )
        self.assertEqual(result["total_input_tokens"], 0)

    @pytest.mark.asyncio