fmt

Author: jackalcooper

tests/mock_server.py

@@ -1265,10 +1265,11 @@ async def dynamic_path_generation(
 
         try:
             payload = await request.json()
-            payload["model"] = model_path
-            body = GenerationRequest(**payload)
         except Exception as e:
-            raise HTTPException(status_code=400, detail=f"Invalid Request: {e}")
+            raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}")
+
+        payload["model"] = model_path
+        body = GenerationRequest(**payload)
 
         accept_header = request.headers.get("accept", "")
         dashscope_sse = request.headers.get("x-dashscope-sse", "").lower()

tests/test_doc.py

@@ -49,14 +49,17 @@
 
 # --- HELPERS ---
 
+
 @dataclass
 class SSEFrame:
     """Formal representation of a Server-Sent Event frame for validation."""
+
     id: str
     output: Dict[str, Any]
     usage: Dict[str, Any]
     request_id: str
 
+
 def parse_sse_stream(response: requests.Response) -> Generator[SSEFrame, None, None]:
     """Parses the raw SSE stream."""
     for line in response.iter_lines():
@@ -67,14 +70,17 @@ def parse_sse_stream(response: requests.Response) -> Generator[SSEFrame, None, N
                 try:
                     data = json.loads(json_str)
                     yield SSEFrame(
-                        id=data.get("output", {}).get("choices", [{}])[0].get("id", "unknown"),
+                        id=data.get("output", {})
+                        .get("choices", [{}])[0]
+                        .get("id", "unknown"),
                         output=data.get("output", {}),
                         usage=data.get("usage", {}),
                         request_id=data.get("request_id", ""),
                     )
                 except json.JSONDecodeError:
                     continue
 
+
 def make_request(payload: Dict[str, Any]):
     """
     Helper to send POST request using the Dynamic Path URL.
@@ -85,7 +91,9 @@ def make_request(payload: Dict[str, Any]):
     model_path = payload.get("model")
 
     if not model_path:
-        raise ValueError("Payload must contain 'model' field for dynamic URL construction")
+        raise ValueError(
+            "Payload must contain 'model' field for dynamic URL construction"
+        )
 
     # 构建动态 URL
     # 例如: http://localhost:8000/siliconflow/models/pre-siliconflow/deepseek-v3
@@ -97,6 +105,7 @@ def make_request(payload: Dict[str, Any]):
 
 # --- TEST SUITE ---
 
+
 class TestParameterValidation:
     """
     对应表格中参数校验相关的错误用例 (4xx Error Codes)
@@ -115,11 +124,15 @@ def test_invalid_parameter_type_top_p(self):
         response = make_request(payload)
 
         # 验证状态码不应为 500
-        assert response.status_code != 500, "Should not return 500 for invalid parameter type"
+        assert (
+            response.status_code != 500
+        ), "Should not return 500 for invalid parameter type"
         assert response.status_code == 400
 
         data = response.json()
-        assert "InvalidParameter" in data.get("code", "") or "InvalidParameter" in data.get("message", "")
+        assert "InvalidParameter" in data.get(
+            "code", ""
+        ) or "InvalidParameter" in data.get("message", "")
 
     @pytest.mark.parametrize("top_p_value", [0, 0.0])
     def test_invalid_parameter_range_top_p(self, top_p_value):
@@ -184,7 +197,9 @@ def test_r1_usage_structure(self):
         # 验证 output_tokens_details 存在
         assert output_details, "output_tokens_details missing"
         # 验证不包含 text_tokens (根据表格描述这是预期行为)
-        assert "text_tokens" not in output_details, "R1 usage should not contain text_tokens"
+        assert (
+            "text_tokens" not in output_details
+        ), "R1 usage should not contain text_tokens"
         # 验证包含 reasoning_tokens
         assert "reasoning_tokens" in output_details
 
@@ -230,7 +245,9 @@ def test_prefix_completion_thinking_conflict(self):
 
         assert response.status_code == 400
         data = response.json()
-        assert "Partial mode is not supported when enable_thinking is true" in data.get("message", "")
+        assert "Partial mode is not supported when enable_thinking is true" in data.get(
+            "message", ""
+        )
 
     def test_history_with_tool_calls(self):
         """
@@ -271,7 +288,9 @@ def test_history_with_tool_calls(self):
         response = make_request(payload)
 
         # 核心验证：不能崩 (500)
-        assert response.status_code != 500, "Server returned 500 for history with tool calls"
+        assert (
+            response.status_code != 500
+        ), "Server returned 500 for history with tool calls"
         assert response.status_code == 200
 
     def test_r1_tool_call_format_wrapping(self):