273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""tests/paper/test_paper_tools.py —— 论文服务工具逻辑单元测试（假 LLM + 假检索）。
|
||
|
||
覆盖：propose_paper_outline / write_paper_section（含引用校验重试）/ get_paper_progress /
|
||
revise_paper_section / revise_paper / get_paper / complete_paper（重排 + disclaimer）/ cancel_paper +
|
||
引用协议纯函数（GB/T 7714 渲染、[n] 重排）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from conftest import _run
|
||
from paper_writer_service.app import db as dbmod
|
||
from paper_writer_service.app.services.paper import cite, complete, outline, write
|
||
|
||
OUTLINE_RES = {
|
||
"title": "水稻SPL基因家族研究进展",
|
||
"abstract_expect": "综述水稻SPL基因家族的结构、功能与调控机制。",
|
||
"sections": [
|
||
{"section_id": "s1", "title": "引言", "key_points": ["SPL家族概述"], "target_words": 1500},
|
||
{"section_id": "s2", "title": "SPL家族结构与进化", "key_points": ["结构特征"], "target_words": 2000},
|
||
{"section_id": "s3", "title": "SPL功能与调控", "key_points": ["功能"], "target_words": 2000},
|
||
{"section_id": "s4", "title": "讨论与展望", "key_points": ["展望"], "target_words": 1500},
|
||
{"section_id": "s5", "title": "结论", "key_points": ["总结"], "target_words": 1000},
|
||
],
|
||
}
|
||
|
||
|
||
def _propose(fake_llm, fake_search):
|
||
fake_llm.responses.append(dict(OUTLINE_RES))
|
||
return _run(outline.propose_paper_outline("水稻SPL基因家族", paper_type="综述", language="中文"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# propose_paper_outline
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_propose_outline_creates_paper_and_sections(paper_db, fake_llm, fake_search):
|
||
r = _propose(fake_llm, fake_search)
|
||
assert r["ok"] is True
|
||
assert r["title"] == "水稻SPL基因家族研究进展"
|
||
assert len(r["sections"]) == 5
|
||
|
||
papers = _run(dbmod.db.fetchall("SELECT * FROM papers"))
|
||
assert len(papers) == 1
|
||
secs = _run(dbmod.db.fetchall("SELECT * FROM paper_sections ORDER BY rowid"))
|
||
assert [s["section_id"] for s in secs] == ["s1", "s2", "s3", "s4", "s5"]
|
||
assert all(s["status"] == "pending" for s in secs)
|
||
|
||
|
||
def test_propose_outline_llm_error(paper_db, fake_llm, fake_search):
|
||
fake_llm.error = "deepseek 超时"
|
||
r = _run(outline.propose_paper_outline("水稻SPL"))
|
||
assert r["ok"] is False
|
||
assert r["code"] == "LLM_ERROR"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# write_paper_section（引用校验 + 落库）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_write_section_creates_citations(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({
|
||
"markdown": "SPL 是植物特有的转录因子家族[1]，水稻有 19 个成员[2]。",
|
||
"citations_used": [
|
||
{"temp_n": 1, "title": "水稻SPL基因家族研究进展", "journal": "植物学报", "year": "2024",
|
||
"doi": "", "pmid": "", "source_doc_id": "d1", "snippet": "x"},
|
||
{"temp_n": 2, "title": "植物SPL转录因子功能", "journal": "", "year": "2023",
|
||
"doi": "", "pmid": "", "source_doc_id": "d2", "snippet": "y"},
|
||
],
|
||
})
|
||
r = _run(write.write_paper_section(paper_id, "s1"))
|
||
assert r["ok"] is True
|
||
assert r["status"] == "done"
|
||
assert len(r["new_citations"]) == 2
|
||
|
||
cits = _run(dbmod.db.fetchall("SELECT * FROM citations ORDER BY temp_n"))
|
||
assert [c["temp_n"] for c in cits] == [1, 2]
|
||
assert cits[0]["kind"] == "title"
|
||
sec = _run(dbmod.db.fetchone("SELECT * FROM paper_sections WHERE paper_id = ? AND section_id = 's1'", (paper_id,)))
|
||
assert sec["status"] == "done"
|
||
assert "[1]" in sec["markdown"]
|
||
|
||
|
||
def test_write_section_drops_fabricated_doi(paper_db, fake_llm, fake_search):
|
||
"""杜撰的 doi/pmid/authors/journal 应被丢弃，强制 title kind 降级格式。"""
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({
|
||
"markdown": "正文[1]。",
|
||
"citations_used": [{"temp_n": 1, "title": "水稻SPL研究", "authors": "X", "journal": "J",
|
||
"year": "2020", "doi": "10.1234/fake", "pmid": "123", "source_doc_id": "d1"}],
|
||
})
|
||
r = _run(write.write_paper_section(paper_id, "s1"))
|
||
assert r["ok"] is True
|
||
cit = _run(dbmod.db.fetchone("SELECT * FROM citations WHERE temp_n = 1"))
|
||
assert cit["kind"] == "title"
|
||
assert "知识库资料" in cit["ref_text"]
|
||
assert "10.1234" not in cit["ref_text"]
|
||
|
||
|
||
def test_write_section_invalid_citation_retries_once(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.calls.clear() # 重置计数，只统计写节调用（大纲已消费 1 次）
|
||
# 第一次引用非法编号 [99]，第二次合法
|
||
fake_llm.responses.append({
|
||
"markdown": "SPL 家族[99]。",
|
||
"citations_used": [{"temp_n": 1, "title": "t", "source_doc_id": "d1"}],
|
||
})
|
||
fake_llm.responses.append({
|
||
"markdown": "SPL 家族[1]。",
|
||
"citations_used": [{"temp_n": 1, "title": "t", "source_doc_id": "d1"}],
|
||
})
|
||
r = _run(write.write_paper_section(paper_id, "s1"))
|
||
assert r["ok"] is True
|
||
assert len(fake_llm.calls) == 2 # 一次失败 + 一次重试成功
|
||
|
||
|
||
def test_write_section_invalid_citation_fails_after_retries(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({"markdown": "SPL 家族[99]。", "citations_used": []})
|
||
fake_llm.responses.append({"markdown": "SPL 家族[88]。", "citations_used": []})
|
||
r = _run(write.write_paper_section(paper_id, "s1"))
|
||
assert r["ok"] is False
|
||
assert r["code"] == "CITE_VALIDATION_FAILED"
|
||
|
||
|
||
def test_write_section_not_found(paper_db, fake_llm):
|
||
r = _run(write.write_paper_section("nope", "s1"))
|
||
assert r["ok"] is False
|
||
assert r["code"] == "NOT_FOUND"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_paper_progress / get_paper
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_progress_reflects_done_section(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({
|
||
"markdown": "正文[1]。",
|
||
"citations_used": [{"temp_n": 1, "title": "t", "source_doc_id": "d1"}],
|
||
})
|
||
_run(write.write_paper_section(paper_id, "s1"))
|
||
r = _run(complete.get_paper_progress(paper_id))
|
||
assert r["ok"] is True
|
||
assert r["status"] == "writing"
|
||
s1 = next(s for s in r["sections"] if s["section_id"] == "s1")
|
||
assert s1["status"] == "done"
|
||
assert s1["citation_count"] == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# revise_paper_section / revise_paper
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_revise_section_appends_new_citation(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({
|
||
"markdown": "旧正文[1]。",
|
||
"citations_used": [{"temp_n": 1, "title": "旧文献", "source_doc_id": "d1"}],
|
||
})
|
||
_run(write.write_paper_section(paper_id, "s1"))
|
||
# 修订：新增 [2]
|
||
fake_llm.responses.append({
|
||
"markdown": "新正文[1][2]。",
|
||
"citations_used": [
|
||
{"temp_n": 1, "title": "旧文献", "source_doc_id": "d1"},
|
||
{"temp_n": 2, "title": "新文献", "source_doc_id": "d3"},
|
||
],
|
||
})
|
||
r = _run(write.revise_paper_section(paper_id, "s1", "更严谨"))
|
||
assert r["ok"] is True
|
||
assert r["citation_delta"] == 1
|
||
cits = _run(dbmod.db.fetchall("SELECT temp_n FROM citations ORDER BY temp_n"))
|
||
assert [c["temp_n"] for c in cits] == [1, 2]
|
||
|
||
|
||
def test_revise_paper_returns_affected(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({"affected_section_ids": ["s2", "s3"], "plan": "修订讨论与结论"})
|
||
r = _run(write.revise_paper(paper_id, "把讨论改严谨"))
|
||
assert r["ok"] is True
|
||
assert r["affected_section_ids"] == ["s2", "s3"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# complete_paper（重排 + GB/T 7714 + disclaimer）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_complete_renumbers_by_first_appearance(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
# s1 先引用 [2]，s2 后引用 [1] → complete 后 [2] 应重排为 [1]
|
||
fake_llm.responses.append({
|
||
"markdown": "引言提到 B[2]。",
|
||
"citations_used": [{"temp_n": 2, "title": "文献B", "source_doc_id": "d2"}],
|
||
})
|
||
_run(write.write_paper_section(paper_id, "s1"))
|
||
fake_llm.responses.append({
|
||
"markdown": "正文提到 A[1]。",
|
||
"citations_used": [{"temp_n": 1, "title": "文献A", "source_doc_id": "d1"}],
|
||
})
|
||
_run(write.write_paper_section(paper_id, "s2"))
|
||
|
||
r = _run(complete.complete_paper(paper_id))
|
||
assert r["ok"] is True
|
||
# 首次出现顺序：B(temp 2) 在 s1，A(temp 1) 在 s2 → 2→1, 1→2
|
||
assert r["renumber_map"] == {2: 1, 1: 2}
|
||
assert r["references"][0]["n"] == 1
|
||
assert r["references"][0]["text"].startswith("文献B")
|
||
assert r["disclaimer"] != "" # title kind → 降级 disclaimer
|
||
|
||
paper = _run(dbmod.db.fetchone("SELECT * FROM papers WHERE id = ?", (paper_id,)))
|
||
assert paper["status"] == "complete"
|
||
assert "[1]" in paper["final_markdown"] and "[2]" in paper["final_markdown"]
|
||
|
||
|
||
def test_complete_paper_no_citations_no_disclaimer(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({"markdown": "无引用正文。", "citations_used": []})
|
||
_run(write.write_paper_section(paper_id, "s1"))
|
||
r = _run(complete.complete_paper(paper_id))
|
||
assert r["ok"] is True
|
||
assert r["references"] == []
|
||
assert r["disclaimer"] == ""
|
||
|
||
|
||
def test_get_paper_returns_full_markdown(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
fake_llm.responses.append({"markdown": "正文[1]。", "citations_used": [
|
||
{"temp_n": 1, "title": "t", "source_doc_id": "d1"}]})
|
||
_run(write.write_paper_section(paper_id, "s1"))
|
||
_run(complete.complete_paper(paper_id))
|
||
r = _run(complete.get_paper(paper_id))
|
||
assert r["ok"] is True
|
||
assert r["status"] == "complete"
|
||
assert "水稻SPL基因家族研究进展" in r["full_markdown"]
|
||
assert len(r["references"]) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# cancel_paper
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_cancel_paper(paper_db, fake_llm, fake_search):
|
||
paper_id = _propose(fake_llm, fake_search)["paper_id"]
|
||
r = _run(complete.cancel_paper(paper_id))
|
||
assert r["ok"] is True
|
||
paper = _run(dbmod.db.fetchone("SELECT status FROM papers WHERE id = ?", (paper_id,)))
|
||
assert paper["status"] == "cancelled"
|
||
|
||
|
||
def test_cancel_paper_not_found(paper_db):
|
||
r = _run(complete.cancel_paper("nope"))
|
||
assert r["ok"] is False
|
||
assert r["code"] == "NOT_FOUND"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 引用协议纯函数（cite）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_gbt7714_doi_pmid_title():
|
||
assert "DOI:10.1000/x" in cite.render_gbt7714({"title": "T", "journal": "J", "year": "2024", "doi": "10.1000/x"})
|
||
assert "PMID:123" in cite.render_gbt7714({"title": "T", "journal": "J", "year": "2024", "pmid": "123"})
|
||
degraded = cite.render_gbt7714({"title": "知识库标题", "year": "2023"})
|
||
assert degraded.endswith("知识库资料, 2023. [EB/OL].")
|
||
|
||
|
||
def test_renumber_map_first_appearance():
|
||
mapping = cite.renumber_map([1, 2, 3], "见[3]和[1]，以及[2]")
|
||
assert mapping == {3: 1, 1: 2, 2: 3}
|
||
assert cite.replace_numbers("见[3]和[1]", {3: 1, 1: 2}) == "见[1]和[2]"
|