LuminousRuoxi
6e0db821a8
- /api/translate 增加逐句模式（sentences）: 前端拆句（en 为原文精确文本）→ 模型逐句翻译， 数量/顺序校验复用 _count_check（不一致自动重试）；响应 en 原样回填保证对位可靠 - 左栏原文区新增「原文中文对照 →」: 逐句中文展示，命中批注 evidence 的句子 与原文同色（note-rhetoric/repeat/growth/structure）+ 同序号（sup）， 点击对照行可联动高亮原文句与批注卡片 - 对照缓存按段保存并持久化（原文变化/换文书自动失效清空） - 测试: 新增逐句模式 2 个 hermetic 测试（33 passed） - 浏览器实测 18/18 通过（含逐句渲染/高亮对位/点击联动/刷新持久化）
605 lines
22 KiB
Python
605 lines
22 KiB
Python
"""Hermetic tests for Human Voice Rewrite Demo (PRD v1.1).
|
||
|
||
No real model calls — the LLM seam is stubbed.
|
||
"""
|
||
|
||
import json
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from evidence import sanitize_diagnosis, sanitize_patterns
|
||
from llm import LLMError, LlmClient, extract_json, message_text, reasoning_config
|
||
from main import app
|
||
from prompts import normalize_diagnose_payload
|
||
from schemas import Annotation, DiagnoseResponse, ParagraphBrief, Pattern
|
||
|
||
SAMPLE_PARAGRAPHS = [
|
||
"The pencil in my hand was still. I wanted to find the answer.",
|
||
"I met a problem that I could not solve right away.",
|
||
"I learned to be patient.",
|
||
"Now, I sit at my desk again.",
|
||
]
|
||
|
||
|
||
class FakeCompleter:
|
||
def __init__(self, payload):
|
||
self.payload = payload
|
||
|
||
def complete_json(self, system, user, validate=None):
|
||
return self.payload
|
||
|
||
|
||
# ---------------------------------------------------------------- evidence
|
||
def test_evidence_keeps_verbatim_hits_drops_misses():
|
||
pats = [
|
||
Pattern(
|
||
pattern_id="P05",
|
||
name="Metaphor Stack",
|
||
affected_paragraphs=["p1"],
|
||
evidence=["the pencil in my hand", "not in the text"],
|
||
why_ai_like="w",
|
||
human_impact="h",
|
||
transformation_rule="Keep One Motif",
|
||
)
|
||
]
|
||
out = sanitize_patterns(pats, SAMPLE_PARAGRAPHS)
|
||
assert out[0].evidence == ["the pencil in my hand"]
|
||
assert out[0].category == "rhetoric"
|
||
|
||
|
||
def test_evidence_unmatched_cleared_but_pattern_kept():
|
||
pats = [
|
||
Pattern(
|
||
pattern_id="P02",
|
||
name="Explicit Lesson",
|
||
affected_paragraphs=["p3"],
|
||
evidence=["somewhere else entirely"],
|
||
why_ai_like="w",
|
||
human_impact="h",
|
||
transformation_rule="Lesson -> Change in Judgment",
|
||
)
|
||
]
|
||
out = sanitize_patterns(pats, SAMPLE_PARAGRAPHS)
|
||
assert len(out) == 1 and out[0].pattern_id == "P02"
|
||
assert out[0].evidence == []
|
||
assert out[0].category == "growth"
|
||
|
||
|
||
def test_fictional_annotation_is_dropped():
|
||
resp = DiagnoseResponse(
|
||
overall_diagnosis="x",
|
||
paragraph_briefs=[
|
||
ParagraphBrief(
|
||
paragraph_id="p1",
|
||
annotations=[
|
||
Annotation(
|
||
pattern_id="P05",
|
||
category="rhetoric",
|
||
title="虚构",
|
||
evidence=["this phrase is not in the paragraph"],
|
||
observation="o",
|
||
rewrite_action="a",
|
||
),
|
||
Annotation(
|
||
pattern_id="P05",
|
||
category="rhetoric",
|
||
title="真实",
|
||
evidence=["the pencil in my hand"],
|
||
observation="o",
|
||
rewrite_action="a",
|
||
),
|
||
],
|
||
)
|
||
],
|
||
)
|
||
cleaned = sanitize_diagnosis(resp, SAMPLE_PARAGRAPHS)
|
||
assert len(cleaned.paragraph_briefs[0].annotations) == 1
|
||
assert cleaned.paragraph_briefs[0].annotations[0].title == "真实"
|
||
|
||
|
||
# ---------------------------------------------------------------- extract_json / retry
|
||
def test_extract_json_strips_fences_and_wraps():
|
||
assert extract_json('```json\n{"a": 1}\n```') == {"a": 1}
|
||
assert extract_json('here you go: {"a": 1} thanks') == {"a": 1}
|
||
|
||
|
||
def test_message_text_joins_content_parts():
|
||
assert message_text({"choices": [{"message": {"content": [{"type": "text", "text": '{"ok":1}'}]}}]}) == '{"ok":1}'
|
||
|
||
|
||
def test_reasoning_config_does_not_set_effort_and_max_tokens():
|
||
first = reasoning_config(False)
|
||
retry = reasoning_config(True)
|
||
assert not ("effort" in first and "max_tokens" in first)
|
||
assert not ("effort" in retry and "max_tokens" in retry)
|
||
assert first == {"max_tokens": 512}
|
||
assert retry == {"effort": "none", "exclude": True}
|
||
|
||
|
||
def test_bad_proxy_url_raises_readable_llm_error():
|
||
# 代理 URL 带变量名前缀（如 .env 解析出错）→ httpx 构造期 ValueError；
|
||
# 必须转可读 LLMError，不能裸 500（2026-08-25 实测事故，见 llm.py 注释）
|
||
c = LlmClient(api_key="test-key", proxy="PRODREAM_BACKEND_OPENROUTER_PROXY_URL=http://x:1@h:2")
|
||
with pytest.raises(LLMError, match="配置错误"):
|
||
c._complete("system", "user")
|
||
|
||
|
||
def test_message_text_empty_mentions_finish_reason():
|
||
with pytest.raises(LLMError, match="finish_reason=length"):
|
||
message_text(
|
||
{
|
||
"choices": [
|
||
{
|
||
"finish_reason": "length",
|
||
"message": {"content": "", "reasoning_content": "thinking..."},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
|
||
|
||
class SeqClient(LlmClient):
|
||
def __init__(self, responses):
|
||
super().__init__(api_key="sk-test", base_url="http://stub", model="stub")
|
||
self.responses = list(responses)
|
||
self.calls = 0
|
||
|
||
def _complete(self, system, user, json_reminder=False, **kwargs):
|
||
self.calls += 1
|
||
kind, val = self.responses.pop(0)
|
||
if kind == "error":
|
||
raise LLMError(val)
|
||
return val
|
||
|
||
|
||
def test_parse_failure_retries_once():
|
||
c = SeqClient([("content", "not json at all"), ("content", '{"ok": 1}')])
|
||
assert c.complete_json("s", "u") == {"ok": 1}
|
||
assert c.calls == 2
|
||
|
||
|
||
def test_transient_5xx_retries_once():
|
||
c = SeqClient([("error", "HTTP 502 bad gateway"), ("content", '{"ok": 2}')])
|
||
assert c.complete_json("s", "u") == {"ok": 2}
|
||
assert c.calls == 2
|
||
|
||
|
||
def test_parse_failure_twice_raises():
|
||
c = SeqClient([("content", "nope"), ("content", "still not json")])
|
||
with pytest.raises(LLMError):
|
||
c.complete_json("s", "u")
|
||
assert c.calls == 2
|
||
|
||
|
||
def test_http_402_error_shows_only_error_message():
|
||
"""HTTP 非 200（如 OpenRouter 402）→ LLMError 只带 error.message，不把完整 JSON body 甩给页面（PRD §11 失败兜底）。"""
|
||
import httpx
|
||
|
||
resp = httpx.Response(
|
||
402,
|
||
json={
|
||
"error": {
|
||
"message": "Insufficient credits. Add more using https://openrouter.ai/settings/credits",
|
||
"code": 402,
|
||
"type": "insufficient_quota",
|
||
}
|
||
},
|
||
request=httpx.Request("POST", "http://stub/chat/completions"),
|
||
)
|
||
|
||
class HttpErrorClient(LlmClient):
|
||
def __init__(self):
|
||
super().__init__(api_key="sk-test", base_url="http://stub", model="stub")
|
||
|
||
def _post(self, payload):
|
||
return resp
|
||
|
||
with pytest.raises(LLMError) as ei:
|
||
HttpErrorClient().complete_json("s", "u")
|
||
msg = str(ei.value)
|
||
assert "Insufficient credits" in msg # error.message 对用户可见
|
||
assert '"error"' not in msg # 不是原始 JSON body
|
||
assert "insufficient_quota" not in msg # 不在 message 里的字段不外泄
|
||
|
||
|
||
def test_no_key_fails_closed(monkeypatch):
|
||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||
monkeypatch.delenv("PRODREAM_BACKEND_OPENROUTER_API_KEY", raising=False)
|
||
with pytest.raises(LLMError):
|
||
LlmClient(api_key="")
|
||
|
||
|
||
def test_normalize_diagnose_payload_maps_legacy_fields():
|
||
data = normalize_diagnose_payload(
|
||
{
|
||
"paragraph_briefs": [
|
||
{
|
||
"paragraph_id": "p1",
|
||
"primary_goal": "收修辞",
|
||
"rewrite_guidance": "不要连续比喻",
|
||
"context_before": "上一段已建立确定感",
|
||
"context_after": "这一段只需写挫败",
|
||
"scaffold": ["What I liked was ______."],
|
||
}
|
||
],
|
||
"patterns": [{"pattern_id": "P05", "name": "Metaphor Stack"}],
|
||
}
|
||
)
|
||
brief = data["paragraph_briefs"][0]
|
||
assert brief["rewrite_goal"] == "收修辞"
|
||
assert brief["ai_focus"] == "不要连续比喻"
|
||
assert "确定感" in brief["context_hint"]
|
||
assert brief["scaffold"] == "What I liked was ______."
|
||
assert data["patterns"][0]["category"] == "rhetoric"
|
||
|
||
|
||
def test_normalize_diagnose_payload_coerces_list_fields():
|
||
"""模型偶发把字符串字段返回成数组（2026-08-21 用户实测 ai_focus 4 段全中）——
|
||
normalize 必须规整为字符串/数组，且规整后能通过 schema（不再 502）。"""
|
||
data = normalize_diagnose_payload(
|
||
{
|
||
"overall_diagnosis": ["开头不错", "结尾乏力"],
|
||
"paragraph_briefs": [
|
||
{
|
||
"paragraph_id": "p1",
|
||
"confirmed_meaning": ["第一段写犹豫"],
|
||
"rewrite_goal": "收修辞",
|
||
"ai_focus": ["重复修辞", "抽象总结"],
|
||
"must_preserve": "secret code",
|
||
"annotations": [{"title": ["短标题"], "observation": "连续两个比喻", "evidence": "原文句"}],
|
||
}
|
||
],
|
||
"patterns": [{"pattern_id": "P05", "name": ["Metaphor", "Stack"]}],
|
||
}
|
||
)
|
||
brief = data["paragraph_briefs"][0]
|
||
assert brief["ai_focus"] == "重复修辞；抽象总结"
|
||
assert brief["confirmed_meaning"] == "第一段写犹豫"
|
||
assert brief["must_preserve"] == ["secret code"]
|
||
assert brief["annotations"][0]["title"] == "短标题"
|
||
assert brief["annotations"][0]["evidence"] == ["原文句"]
|
||
assert data["overall_diagnosis"] == "开头不错；结尾乏力"
|
||
assert data["patterns"][0]["name"] == "Metaphor；Stack"
|
||
# 规整后必须能过 schema——复现用户实测 502 场景不再发生
|
||
resp = DiagnoseResponse.model_validate(data)
|
||
assert resp.paragraph_briefs[0].ai_focus == "重复修辞；抽象总结"
|
||
|
||
|
||
# ---------------------------------------------------------------- API shape
|
||
def api_client(payload):
|
||
import main as main_mod
|
||
|
||
main_mod.get_client = lambda: FakeCompleter(payload)
|
||
return TestClient(app)
|
||
|
||
|
||
class CaptureCompleter:
|
||
def __init__(self, payload):
|
||
self.payload = payload
|
||
self.calls = []
|
||
|
||
def complete_json(self, system, user, validate=None):
|
||
self.calls.append({"system": system, "user": user})
|
||
return self.payload
|
||
|
||
|
||
def test_api_diagnose_passes_constraints_to_llm():
|
||
"""补充改写要求/约束必须真实传给 LLM（诊断重生成链路）。"""
|
||
import main as main_mod
|
||
|
||
cap = CaptureCompleter(
|
||
{
|
||
"overall_diagnosis": "d",
|
||
"patterns": [],
|
||
"paragraph_briefs": [
|
||
{
|
||
"paragraph_id": "p1",
|
||
"confirmed_meaning": "m",
|
||
"rewrite_goal": "g",
|
||
"ai_focus": "a",
|
||
"annotations": [],
|
||
}
|
||
],
|
||
"optional_suggestion": "",
|
||
}
|
||
)
|
||
main_mod.get_client = lambda: cap
|
||
r = TestClient(app).post(
|
||
"/api/diagnose",
|
||
json={
|
||
"paragraphs": SAMPLE_PARAGRAPHS,
|
||
"confirmed_anchors": ["a"],
|
||
"global_constraints": ["整体更直接、克制"],
|
||
"paragraph_constraints": {"p1": ["保留 secret code"]},
|
||
},
|
||
)
|
||
assert r.status_code == 200
|
||
assert "整体更直接、克制" in cap.calls[0]["user"]
|
||
assert "保留 secret code" in cap.calls[0]["user"]
|
||
|
||
|
||
def test_api_diagnose_coerces_ai_focus_list():
|
||
"""复现 2026-08-21 用户实测：模型把 ai_focus 返回成数组——
|
||
路由必须规整为字符串并返回 200，而不是 502。"""
|
||
import main as main_mod
|
||
|
||
cap = CaptureCompleter(
|
||
{
|
||
"overall_diagnosis": ["开头不错", "结尾乏力"],
|
||
"patterns": [],
|
||
"paragraph_briefs": [
|
||
{
|
||
"paragraph_id": f"p{i + 1}",
|
||
"confirmed_meaning": "m",
|
||
"rewrite_goal": "g",
|
||
"ai_focus": ["重复修辞", "抽象总结"],
|
||
"annotations": [],
|
||
}
|
||
for i in range(4)
|
||
],
|
||
"optional_suggestion": "",
|
||
}
|
||
)
|
||
main_mod.get_client = lambda: cap
|
||
r = TestClient(app).post(
|
||
"/api/diagnose",
|
||
json={"paragraphs": SAMPLE_PARAGRAPHS, "confirmed_anchors": [], "global_constraints": []},
|
||
)
|
||
assert r.status_code == 200
|
||
briefs = r.json()["paragraph_briefs"]
|
||
assert len(briefs) == 4
|
||
assert briefs[0]["ai_focus"] == "重复修辞；抽象总结"
|
||
|
||
|
||
def test_api_analyze_shape():
|
||
payload = {
|
||
"essay_summary": "学生和数学关系的变化。",
|
||
"prompt_alignment": {"prompt_intent": "", "current_alignment": "", "optional_opportunity": ""},
|
||
"paragraphs": [
|
||
{
|
||
"id": "p1",
|
||
"original_text": SAMPLE_PARAGRAPHS[0],
|
||
"natural_meaning_zh": "喜欢数学带来的确定感。",
|
||
"semantic_anchor": "数学给我确定感。",
|
||
"optional_content_opportunity": "",
|
||
}
|
||
],
|
||
}
|
||
r = api_client(payload).post(
|
||
"/api/analyze", json={"prompt": "P1", "word_limit": 650, "paragraphs": [SAMPLE_PARAGRAPHS[0]]}
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["paragraphs"][0]["id"] == "p1"
|
||
|
||
|
||
def test_api_analyze_empty_text_422():
|
||
r = api_client({}).post("/api/analyze", json={"paragraphs": [" "]})
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_api_analyze_bad_payload_returns_one_line_detail():
|
||
"""LLM 输出不符合 schema → 502 detail 是一行用户可读摘要，不把 ValidationError 堆栈甩给页面（PRD §11 失败兜底）。"""
|
||
r = api_client({"essay_summary": "s", "paragraphs": [123]}).post( # 类型错误触发 ValidationError
|
||
"/api/analyze", json={"prompt": "P1", "word_limit": 650, "paragraphs": [SAMPLE_PARAGRAPHS[0]]}
|
||
)
|
||
assert r.status_code == 502
|
||
detail = r.json()["detail"]
|
||
assert "结果格式不符合预期" in detail
|
||
assert "validation errors for" not in detail # 没有完整校验堆栈
|
||
assert detail.count("\n") == 0 # 单行
|
||
assert len(detail) < 200 # 摘要级长度
|
||
|
||
|
||
def test_api_diagnose_drops_fictional_evidence():
|
||
payload = {
|
||
"overall_diagnosis": "修辞偏密。",
|
||
"patterns": [
|
||
{
|
||
"pattern_id": "P05",
|
||
"name": "Metaphor Stack",
|
||
"category": "rhetoric",
|
||
"affected_paragraphs": ["p1"],
|
||
"evidence": ["不存在的证据"],
|
||
"why_ai_like": "w",
|
||
"human_impact": "h",
|
||
"transformation_rule": "Keep One Motif",
|
||
}
|
||
],
|
||
"paragraph_briefs": [
|
||
{
|
||
"paragraph_id": "p1",
|
||
"confirmed_meaning": "确定感",
|
||
"rewrite_goal": "收修辞",
|
||
"ai_focus": "连续比喻",
|
||
"annotations": [
|
||
{
|
||
"pattern_id": "P05",
|
||
"category": "rhetoric",
|
||
"kind_label": "修辞包装",
|
||
"title": "虚构",
|
||
"evidence": ["不存在的证据"],
|
||
"observation": "o",
|
||
"rewrite_action": "a",
|
||
}
|
||
],
|
||
"context_hint": "",
|
||
"scaffold": "What I liked was ______.",
|
||
"reference_snippet": "I liked the certainty.",
|
||
}
|
||
],
|
||
"optional_suggestion": "",
|
||
}
|
||
r = api_client(payload).post(
|
||
"/api/diagnose",
|
||
json={"paragraphs": SAMPLE_PARAGRAPHS, "confirmed_anchors": ["a"], "global_constraints": [], "paragraph_constraints": {}},
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["patterns"][0]["pattern_id"] == "P05"
|
||
assert body["patterns"][0]["evidence"] == []
|
||
assert body["paragraph_briefs"][0]["annotations"] == []
|
||
|
||
|
||
def test_api_recheck_revision_single_target():
|
||
payload = {
|
||
"status": "revision_required",
|
||
"checked_rewrite_version": "rv_12",
|
||
"global_checks": {
|
||
k: "pass"
|
||
for k in [
|
||
"semantic_preservation",
|
||
"voice_consistency",
|
||
"pattern_reduction",
|
||
"new_pattern",
|
||
"coherence",
|
||
"reference_copying",
|
||
"word_limit",
|
||
]
|
||
},
|
||
"revision_targets": [
|
||
{
|
||
"paragraph_id": "p3",
|
||
"blocking_issue": "新版本仍是 Explicit Lesson",
|
||
"evidence": ["gradually came to realize"],
|
||
"single_revision_goal": "不要总结我学到了什么，写判断怎么变化。",
|
||
}
|
||
],
|
||
}
|
||
r = api_client(payload).post(
|
||
"/api/recheck",
|
||
json={
|
||
"original_paragraphs": SAMPLE_PARAGRAPHS,
|
||
"rewrite_paragraphs": ["a" * 60] * 4,
|
||
"rewrite_version": "rv_12",
|
||
},
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["status"] == "revision_required"
|
||
assert body["checked_rewrite_version"] == "rv_12"
|
||
assert len(body["revision_targets"]) == 1
|
||
|
||
|
||
def test_api_recheck_empty_paragraph_422():
|
||
r = api_client({}).post(
|
||
"/api/recheck",
|
||
json={"original_paragraphs": SAMPLE_PARAGRAPHS, "rewrite_paragraphs": ["ok", "", "ok", "ok"]},
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_api_reference_shape():
|
||
payload = {"paragraph_id": "p1", "starter": "What I liked about math was ______."}
|
||
r = api_client(payload).post(
|
||
"/api/reference",
|
||
json={"paragraph_id": "p1", "original_text": SAMPLE_PARAGRAPHS[0], "semantic_anchor": "a", "rewrite_goal": "g"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["starter"].startswith("What I liked")
|
||
assert r.json()["reference_snippet"].startswith("What I liked")
|
||
|
||
|
||
def test_api_scaffold_shape():
|
||
payload = {"paragraph_id": "p1", "scaffold": "What I liked about math was ______."}
|
||
r = api_client(payload).post(
|
||
"/api/scaffold",
|
||
json={"paragraph_id": "p1", "original_text": SAMPLE_PARAGRAPHS[0], "semantic_anchor": "a", "rewrite_goal": "g"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert "liked" in r.json()["scaffold"]
|
||
|
||
|
||
def test_api_translate_shape():
|
||
payload = {"translation": "我不断回到这个问题上。"}
|
||
r = api_client(payload).post(
|
||
"/api/translate",
|
||
json={"text": "I kept coming back to the problem.", "paragraph_id": "p3"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["translation"].startswith("我不断回到")
|
||
assert r.json()["paragraph_id"] == "p3"
|
||
|
||
|
||
def test_api_translate_coerces_list_translation():
|
||
"""模型偶发把 translation 返回成数组 —— 归并为一个字符串，不 502。"""
|
||
r = api_client({"translation": ["第一句。", "第二句。"]}).post(
|
||
"/api/translate", json={"text": "One. Two."}
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["translation"] == "第一句。第二句。"
|
||
|
||
|
||
def test_api_translate_sentences_shape():
|
||
"""原文中文对照（逐句）：en 原样回填、zh 按序对位 —— 前端据此做高亮对位。"""
|
||
payload = {"sentences": [{"zh": "第一句译文。"}, {"zh": "第二句译文。"}]}
|
||
r = api_client(payload).post(
|
||
"/api/translate", json={"sentences": ["One.", "Two."], "paragraph_id": "p1"}
|
||
)
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert [s["en"] for s in data["sentences"]] == ["One.", "Two."]
|
||
assert data["sentences"][1]["zh"] == "第二句译文。"
|
||
|
||
|
||
def test_api_translate_sentences_blank_422():
|
||
r = api_client({"sentences": [{"zh": "x"}]}).post(
|
||
"/api/translate", json={"sentences": [" ", ""]}
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_api_translate_empty_translation_502():
|
||
r = api_client({"translation": " "}).post("/api/translate", json={"text": "Hello."})
|
||
assert r.status_code == 502
|
||
assert "翻译结果为空" in r.json()["detail"]
|
||
|
||
|
||
def test_complete_json_paragraph_count_mismatch_retries_once():
|
||
"""语义校验：段落数不匹配视为一次失败并自动重试，重试成功返回完整结果。
|
||
真实 LLM 偶发返回合法 JSON 但段落条目不全（AI 初审"4 段只显示 1 段"），
|
||
此前无校验被静默接受；现在按 complete_json 既有重试约定自动再试一次。"""
|
||
one = {"essay_summary": "s", "paragraphs": [{"id": "p1", "natural_meaning_zh": "第一段"}]}
|
||
four = {"essay_summary": "s", "paragraphs": [{"id": f"p{i}", "natural_meaning_zh": f"第{i}段"} for i in range(1, 5)]}
|
||
c = SeqClient([("content", json.dumps(one)), ("content", json.dumps(four))])
|
||
out = c.complete_json(
|
||
"s", "u", validate=lambda d: "结果段落数与输入不一致" if len(d.get("paragraphs") or []) != 4 else None
|
||
)
|
||
assert len(out["paragraphs"]) == 4
|
||
assert c.calls == 2 # 第一次因数量不匹配被重试
|
||
|
||
|
||
def test_complete_json_paragraph_count_mismatch_twice_raises():
|
||
"""两次都数量不匹配 → LLMError 带一行可读说明（不把原始 JSON 甩给页面）。"""
|
||
one = {"essay_summary": "s", "paragraphs": [{"id": "p1", "natural_meaning_zh": "第一段"}]}
|
||
c = SeqClient([("content", json.dumps(one)), ("content", json.dumps(one))])
|
||
with pytest.raises(LLMError) as ei:
|
||
c.complete_json(
|
||
"s", "u",
|
||
validate=lambda d: "结果段落数与输入不一致（输入 4 段，返回 1 段）" if len(d.get("paragraphs") or []) != 4 else None,
|
||
)
|
||
assert "输入 4 段" in str(ei.value) and "返回 1 段" in str(ei.value)
|
||
assert c.calls == 2
|
||
|
||
|
||
def test_api_analyze_wires_paragraph_count_check():
|
||
"""analyze 路由必须把段落数量校验传给 complete_json（否则 4 段输入只返回 1 段会被静默接受）。"""
|
||
import main as main_mod
|
||
|
||
class SpyCompleter(CaptureCompleter):
|
||
def complete_json(self, system, user, validate=None):
|
||
self.validate = validate
|
||
return super().complete_json(system, user, validate)
|
||
|
||
payload = {"essay_summary": "s", "paragraphs": [{"id": "p1", "natural_meaning_zh": "第一段"}]}
|
||
spy = SpyCompleter(payload)
|
||
main_mod.get_client = lambda: spy
|
||
r = TestClient(app).post(
|
||
"/api/analyze", json={"prompt": "P1", "word_limit": 650, "paragraphs": SAMPLE_PARAGRAPHS}
|
||
)
|
||
assert r.status_code == 200
|
||
assert spy.validate is not None
|
||
bad = spy.validate({"essay_summary": "s", "paragraphs": [{"id": "p1"}]})
|
||
assert bad is not None and "段落数" in bad
|
||
assert spy.validate({"essay_summary": "s", "paragraphs": [{"id": f"p{i}"} for i in range(1, 5)]}) is None
|