"""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]"