|
42 | 42 |
|
43 | 43 | # --- HELPERS --- |
44 | 44 |
|
45 | | -def make_request(payload: Dict[str, Any]) -> requests.Response: |
46 | | - """Helper to send POST request using the Dynamic Path URL structure.""" |
| 45 | +def make_request(payload: Dict[str, Any], stream: bool = True) -> requests.Response: |
| 46 | + """ |
| 47 | + Helper to send POST request using the Dynamic Path URL structure. |
| 48 | + Args: |
| 49 | + payload: The request body. |
| 50 | + stream: Whether to use streaming (SSE). Defaults to True. |
| 51 | + """ |
47 | 52 | raw_model_name = payload.get("model") |
48 | 53 |
|
49 | | - # [MODIFIED] 严格检查:如果模型名称不在映射中,直接报错(Fail Test) |
50 | 54 | if raw_model_name not in MODEL_PATH_MAP: |
51 | | - pytest.fail(f"Configuration Error: Model '{raw_model_name}' is not defined in MODEL_PATH_MAP. Please update the mapping.") |
| 55 | + pytest.fail(f"Configuration Error: Model '{raw_model_name}' not in MAP.") |
52 | 56 |
|
53 | 57 | model_path = MODEL_PATH_MAP[raw_model_name] |
54 | | - |
55 | | - # [注释] 测试的时候必须是http://localhost:8000/siliconflow/models/deepseek-ai/DeepSeek-V3这样的URL |
56 | 58 | url = f"{BASE_URL_PREFIX}/{model_path}" |
57 | 59 |
|
58 | | - return requests.post(url, headers=HEADERS, json=payload, stream=True) |
| 60 | + # 处理 Headers |
| 61 | + current_headers = HEADERS.copy() |
| 62 | + if not stream: |
| 63 | + current_headers["Accept"] = "application/json" |
| 64 | + # 移除 SSE 标记 |
| 65 | + if "X-DashScope-SSE" in current_headers: |
| 66 | + del current_headers["X-DashScope-SSE"] |
| 67 | + |
| 68 | + # 注意:requests.post 的 stream 参数控制是否立即下载响应体, |
| 69 | + # 但业务层面的流式由 headers 和 json body 决定,这里保持 requests 的 stream=True |
| 70 | + # 只是为了方便拿到原始响应,或者统一设为 stream 变量也可以。 |
| 71 | + return requests.post(url, headers=current_headers, json=payload, stream=True) |
59 | 72 |
|
60 | 73 |
|
61 | 74 | def assert_exact_error( |
@@ -199,36 +212,45 @@ def test_stop_parameter_invalid_type(self): |
199 | 212 |
|
200 | 213 | def test_non_streaming_request(self): |
201 | 214 | """ |
202 | | - [Updated] 验证非流式返回,需同样应用 URL 映射逻辑和严格检查 |
| 215 | + [Row 16] 验证非流式返回 (incremental_output=false) |
| 216 | + 使用优化后的 make_request,代码更干净 |
203 | 217 | """ |
204 | 218 | payload = { |
205 | 219 | "model": "pre-siliconflow/deepseek-v3.2", |
206 | 220 | "input": {"messages": [{"role": "user", "content": "Say hi"}]}, |
207 | | - "parameters": {}, |
| 221 | + "parameters": {"incremental_output": False}, # 显式设置参数 |
208 | 222 | } |
209 | 223 |
|
210 | | - # 1. 解析模型并应用严格映射检查 |
211 | | - raw_model_name = payload.get("model") |
212 | | - |
213 | | - # [MODIFIED] 严格检查 |
214 | | - if raw_model_name not in MODEL_PATH_MAP: |
215 | | - pytest.fail(f"Configuration Error: Model '{raw_model_name}' is not defined in MODEL_PATH_MAP. Please update the mapping.") |
216 | | - |
217 | | - model_path = MODEL_PATH_MAP[raw_model_name] |
218 | | - |
219 | | - # 2. 拼接 URL |
220 | | - url = f"{BASE_URL_PREFIX}/{model_path}" |
221 | | - |
222 | | - headers_no_stream = HEADERS.copy() |
223 | | - headers_no_stream["Accept"] = "application/json" |
224 | | - del headers_no_stream["X-DashScope-SSE"] |
225 | | - |
226 | | - response = requests.post(url, headers=headers_no_stream, json=payload) |
| 224 | + # 调用优化后的 helper,传入 stream=False |
| 225 | + response = make_request(payload, stream=False) |
227 | 226 |
|
228 | 227 | assert response.status_code == 200, f"Non-streaming request failed: {response.text}" |
229 | 228 | data = response.json() |
| 229 | + # 非流式应该直接返回 output 字段,而不是 SSE 的 delta |
230 | 230 | assert "output" in data, "Non-streaming response missing 'output' field" |
231 | 231 |
|
| 232 | + def test_stop_behavior_in_reasoning(self): |
| 233 | + """ |
| 234 | + [Row 7] stop 在 reasoning content 阶段生效 |
| 235 | + 验证:设置 stop word 为思考过程中必然出现的词(如 'step' 或 'thinking'), |
| 236 | + 确认模型是否异常截断或按预期行为处理(取决于预期是忽略还是截断)。 |
| 237 | + """ |
| 238 | + payload = { |
| 239 | + "model": "pre-siliconflow/deepseek-r1", # 必须是思考模型 |
| 240 | + "input": {"messages": [{"role": "user", "content": "Count from 1 to 5 step by step"}]}, |
| 241 | + "parameters": { |
| 242 | + "stop": ["step"], # 这是一个在思考过程中很容易出现的词 |
| 243 | + "enable_thinking": True |
| 244 | + } |
| 245 | + } |
| 246 | + response = make_request(payload) |
| 247 | + |
| 248 | + # 这里的断言取决于你们的预期: |
| 249 | + # 如果预期是不支持(即不会截断),则 status_code 200 且内容完整。 |
| 250 | + # 如果预期是会截断导致格式错误,则捕获错误。 |
| 251 | + assert response.status_code == 200 |
| 252 | + # 进一步可以检查 response.text 中是否包含 reasoning_content |
| 253 | + |
232 | 254 | def test_thinking_budget_behavior(self): |
233 | 255 | payload = { |
234 | 256 | "model": "pre-siliconflow/deepseek-v3.1", |
|
0 commit comments