297 lines
14 KiB
Python
297 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""结题初稿内核金标准用例（离线、确定性；docx 用 zipfile 断言内容）。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import unittest
|
||
import zipfile
|
||
|
||
from _support import IsolatedDataRoot
|
||
|
||
|
||
def _project(**overrides) -> dict:
|
||
base = {
|
||
"projectNo": "NSFC2026003",
|
||
"projectName": "面向能源转化的材料设计",
|
||
"piName": "王教授",
|
||
"department": "化材学院",
|
||
"program": "国家自然科学基金青年项目",
|
||
"startDate": "2024-01-01",
|
||
"endDate": "2026-12-31",
|
||
"expected": [
|
||
{"type": "论文", "target": 3, "unit": "篇"},
|
||
{"type": "专利", "target": 1, "unit": "项"},
|
||
{"type": "人才培养", "target": 2, "unit": "名"},
|
||
],
|
||
"annualReportExcerpt": "完成界面改性工艺探索；发表高水平论文两篇。",
|
||
"status": "待结题",
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
def _achievements() -> list[dict]:
|
||
return [
|
||
{"projectNo": "NSFC2026003", "achievementType": "论文", "title": "界面调控研究A", "venueOrNo": "Applied Energy", "achDate": "2025-06-01", "isRepresentative": "1"},
|
||
{"projectNo": "NSFC2026003", "achievementType": "论文", "title": "界面调控研究B", "venueOrNo": "J. Mater. Chem. A", "achDate": "2025-11-01", "isRepresentative": "1"},
|
||
{"projectNo": "NSFC2026003", "achievementType": "论文", "title": "界面调控研究C", "venueOrNo": "", "achDate": "2026-03-01", "isRepresentative": "0"},
|
||
{"projectNo": "NSFC2026003", "achievementType": "论文", "title": "界面调控研究D", "venueOrNo": "", "achDate": "2026-05-01", "isRepresentative": "0"},
|
||
{"projectNo": "NSFC2026003", "achievementType": "专利", "title": "一种界面改性方法", "venueOrNo": "ZL2025XXXX", "achDate": "2026-01-01", "isRepresentative": "0"},
|
||
{"projectNo": "NSFC2026003", "achievementType": "人才培养", "title": "硕士研究生1名毕业", "venueOrNo": "", "achDate": "2026-06-30", "isRepresentative": "0"},
|
||
]
|
||
|
||
|
||
def _expenses() -> list[dict]:
|
||
return [
|
||
{"projectNo": "NSFC2026003", "category": "差旅费", "amount": 23000.0, "occurredDate": "2025-04-10", "description": "赴外地调研"},
|
||
{"projectNo": "NSFC2026003", "category": "差旅费", "amount": 10000.0, "occurredDate": "2025-09-01", "description": "学术会议"},
|
||
{"projectNo": "NSFC2026003", "category": "设备费", "amount": 200000.0, "occurredDate": "2024-12-01", "description": "测试设备"},
|
||
{"projectNo": "NSFC2026003", "category": "劳务费", "amount": 40000.0, "occurredDate": "2026-06-01", "description": "研究生劳务"},
|
||
]
|
||
|
||
|
||
class DraftKernelTestBase(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self._root = IsolatedDataRoot()
|
||
self._root.__enter__()
|
||
from db.connection import init_db
|
||
|
||
init_db()
|
||
|
||
def tearDown(self) -> None:
|
||
self._root.__exit__(None, None, None)
|
||
|
||
def _kernel(self, expected=None, achievements=None, expenses=None):
|
||
from service.draft_service import build_draft_sections
|
||
|
||
project = _project()
|
||
if expected is not None:
|
||
project["expected_json"] = json.dumps(expected, ensure_ascii=False)
|
||
return build_draft_sections(project, achievements if achievements is not None else _achievements(), expenses if expenses is not None else _expenses())
|
||
|
||
|
||
class TestC1Completion(DraftKernelTestBase):
|
||
def test_over_and_equal(self) -> None:
|
||
result = self._kernel()
|
||
by_type = {row["type"]: row for row in result["completion"]}
|
||
self.assertEqual(by_type["论文"]["verdict"], "超额完成") # 4 > 3
|
||
self.assertEqual(by_type["专利"]["verdict"], "持平") # 1 == 1
|
||
|
||
def test_below_target(self) -> None:
|
||
result = self._kernel()
|
||
by_type = {row["type"]: row for row in result["completion"]}
|
||
self.assertEqual(by_type["人才培养"]["verdict"], "未达标") # 1 < 2
|
||
self.assertIn("预期 2 名，实际 1 名", by_type["人才培养"]["detail"])
|
||
|
||
def test_missing_type_is_not_met_with_zero(self) -> None:
|
||
result = self._kernel()
|
||
by_type = {row["type"]: row for row in result["completion"]}
|
||
self.assertEqual(by_type["专利"]["actual"], 1)
|
||
|
||
def test_missing_target_is_pending(self) -> None:
|
||
result = self._kernel(expected=[{"type": "论文", "unit": "篇"}])
|
||
self.assertEqual(result["completion"][0]["verdict"], "数据待补")
|
||
|
||
def test_empty_ledger_is_pending_not_unmet(self) -> None:
|
||
"""成果台账整表为空 → 全部"数据待补"，不下"未达标"结论（C1/W1）。"""
|
||
result = self._kernel(achievements=[])
|
||
for row in result["completion"]:
|
||
self.assertEqual(row["verdict"], "数据待补")
|
||
self.assertIn("尚未录入", row["detail"])
|
||
|
||
def test_present_ledger_missing_type_is_unmet_zero(self) -> None:
|
||
"""台账非空但某类型无记录 → 实际按 0 计，未达标（事实陈述）。"""
|
||
result = self._kernel(achievements=[ach for ach in _achievements() if ach["achievementType"] != "专利"])
|
||
by_type = {row["type"]: row for row in result["completion"]}
|
||
self.assertEqual(by_type["专利"]["verdict"], "未达标")
|
||
self.assertEqual(by_type["专利"]["actual"], 0)
|
||
self.assertIn("实际 0 项", by_type["专利"]["detail"])
|
||
|
||
|
||
class TestC3Expenses(DraftKernelTestBase):
|
||
def test_ratio_sums_to_one_and_sentences(self) -> None:
|
||
result = self._kernel()
|
||
total_ratio = round(sum(row["ratio"] for row in result["expense_rows"]), 4)
|
||
self.assertEqual(total_ratio, 1.0)
|
||
top = result["expense_rows"][0]
|
||
self.assertEqual(top["category"], "设备费") # 金额最大排前
|
||
self.assertIn("购置/使用科研设备", top["sentence"])
|
||
|
||
def test_empty_expenses_placeholder(self) -> None:
|
||
result = self._kernel(expenses=[])
|
||
self.assertIn("尚未录入", result["expense_paragraph"])
|
||
|
||
def test_balance_within_approved_budget(self) -> None:
|
||
# 台账支出 273000 元，立项经费 500000 元 → 结余 22.70 万元
|
||
result = self._kernel()
|
||
total = round(sum(row["amount"] for row in result["expense_rows"]), 2)
|
||
self.assertEqual(total, 273000.0)
|
||
with_approved = self._kernel_with_approved(500000)
|
||
self.assertIn("合计支出 27.30 万元", with_approved)
|
||
self.assertIn("结余 22.70 万元", with_approved)
|
||
self.assertIn("范围内", with_approved)
|
||
|
||
def test_overspend_flags_exceed(self) -> None:
|
||
paragraph = self._kernel_with_approved(200000)
|
||
self.assertIn("超出立项经费", paragraph)
|
||
self.assertIn("7.30 万元", paragraph)
|
||
|
||
def _kernel_with_approved(self, approved: float) -> str:
|
||
from service.draft_service import build_draft_sections
|
||
|
||
project = _project(approvedBudget=approved)
|
||
return build_draft_sections(project, _achievements(), _expenses())["expense_paragraph"]
|
||
|
||
|
||
class TestC4AndC5(DraftKernelTestBase):
|
||
def test_narrative_points_with_source(self) -> None:
|
||
result = self._kernel()
|
||
sources = {p["source"] for p in result["narrative_points"]}
|
||
self.assertIn("成果清单", sources)
|
||
self.assertIn("年度报告", sources)
|
||
|
||
def test_sections_order_and_placeholders(self) -> None:
|
||
result = self._kernel()
|
||
titles = list(result["sections"].keys())
|
||
self.assertEqual(titles[:3], ["1. 研究背景与目标", "2. 研究内容与进展", "3. 预期成果完成情况"])
|
||
self.assertIn("来源：成果清单", "\n".join(result["sections"]["5. 成果清单"]))
|
||
self.assertTrue(any(p.startswith("[待补充") for p in result["sections"]["1. 研究背景与目标"]))
|
||
|
||
def test_w1_empty_ledger_does_not_block(self) -> None:
|
||
result = self._kernel(achievements=[], expenses=[])
|
||
self.assertIn("尚未录入", "\n".join(result["sections"]["5. 成果清单"]))
|
||
self.assertIn("尚未录入", result["expense_paragraph"])
|
||
|
||
|
||
class TestDraftGeneration(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self._root = IsolatedDataRoot()
|
||
self._root.__enter__()
|
||
from db.connection import init_db
|
||
from service.ledger_service import import_expenses, import_achievements, import_projects, upsert_project
|
||
|
||
init_db()
|
||
upsert_project(_project())
|
||
for ach in _achievements():
|
||
from service.ledger_service import upsert_achievement
|
||
|
||
upsert_achievement(ach)
|
||
for expense in _expenses():
|
||
from service.ledger_service import upsert_expense
|
||
|
||
upsert_expense(expense)
|
||
|
||
def tearDown(self) -> None:
|
||
self._root.__exit__(None, None, None)
|
||
|
||
def _draft(self, project_no="NSFC2026003"):
|
||
from service.draft_service import query_pending_projects, run_draft
|
||
from service.ledger_service import list_achievements, list_expenses
|
||
|
||
projects = query_pending_projects(project_no=project_no)
|
||
assert projects, "project should exist"
|
||
return run_draft(projects[0], list_achievements(project_no), list_expenses(project_no))
|
||
|
||
def test_docx_generated_with_sections(self) -> None:
|
||
result = self._draft()
|
||
self.assertTrue(result["draft_file"].endswith(".docx"))
|
||
with zipfile.ZipFile(result["draft_file"]) as zf:
|
||
xml = zf.read("word/document.xml").decode("utf-8")
|
||
self.assertIn("结题报告（初稿）", xml)
|
||
self.assertIn("研究背景与目标", xml)
|
||
self.assertIn("经费决算说明", xml)
|
||
self.assertIn("来源：成果清单", xml)
|
||
self.assertIn("待补充", xml)
|
||
self.assertIn('w:highlight w:val="yellow"', xml) # [待补充] 占位黄底提示
|
||
|
||
def test_preview_named_by_draft_id_and_status_update(self) -> None:
|
||
result = self._draft()
|
||
self.assertTrue(result["preview_file"].endswith("draft-1.html"))
|
||
from service.ledger_service import list_projects
|
||
|
||
row = next(p for p in list_projects() if p["project_no"] == "NSFC2026003")
|
||
self.assertEqual(row["status"], "初稿已生成")
|
||
|
||
def test_second_draft_preview_not_overwritten(self) -> None:
|
||
from service.ledger_service import upsert_project
|
||
from service.draft_service import query_pending_projects, run_draft
|
||
|
||
upsert_project(_project(projectNo="NSFC2026004", projectName="第二个项目"))
|
||
result2 = run_draft(query_pending_projects(project_no="NSFC2026004")[0], [], [])
|
||
result1 = run_draft(query_pending_projects(project_no="NSFC2026003")[0], [], [])
|
||
self.assertIn("draft-1.html", result2["preview_file"])
|
||
self.assertIn("draft-2.html", result1["preview_file"])
|
||
self.assertNotEqual(result1["preview_file"], result2["preview_file"])
|
||
|
||
def test_rebuild_preview_is_full_sections(self) -> None:
|
||
from service.report_service import build_preview_for_draft_id
|
||
|
||
result = self._draft()
|
||
draft_id = result["draft_id"]
|
||
path = build_preview_for_draft_id(draft_id)
|
||
self.assertIn(f"draft-{draft_id}.html", path)
|
||
with open(path, encoding="utf-8") as f:
|
||
html_text = f.read()
|
||
# 与 Word 初稿同构：决算、成果清单、占位指引齐全（而非降级的三节摘要）
|
||
self.assertIn("经费决算说明", html_text)
|
||
self.assertIn("成果清单", html_text)
|
||
self.assertIn("研究背景与目标", html_text)
|
||
self.assertIn("待补充", html_text)
|
||
|
||
def test_batch_pending(self) -> None:
|
||
from service.draft_service import query_pending_projects, run_draft
|
||
|
||
for project in query_pending_projects():
|
||
run_draft(project, [], [])
|
||
from service.ledger_service import list_drafts, stats
|
||
|
||
payload = stats()
|
||
self.assertEqual(payload["draft_total"], 1)
|
||
self.assertEqual(len(list_drafts()), 1)
|
||
|
||
def test_project_not_found(self) -> None:
|
||
from service.draft_service import query_pending_projects
|
||
|
||
self.assertEqual(query_pending_projects(project_no="NOPE0001"), [])
|
||
|
||
|
||
class TestCliContract(unittest.TestCase):
|
||
"""CLI 契约：文档 §8 的六个台账命令可解析，render-html 占位符容错。"""
|
||
|
||
def _parser(self):
|
||
from cli.app import build_parser
|
||
|
||
return build_parser()
|
||
|
||
def test_contract_commands_parse(self) -> None:
|
||
parser = self._parser()
|
||
cases = [
|
||
["import-projects", "--projects-json", "x.json"],
|
||
["import-achievements", "--achievements-json", "-"],
|
||
["import-expenses", "--expenses-json", "x.json"],
|
||
["list-projects", "--status", "待结题"],
|
||
["list-achievements", "--project-no", "NSFC2026003"],
|
||
["list-expenses", "--project-no", "NSFC2026003"],
|
||
["add-project", "--project-no", "X", "--approved-budget", "500000"],
|
||
]
|
||
for argv in cases:
|
||
args = parser.parse_args(argv)
|
||
self.assertTrue(callable(getattr(args, "handler", None)), argv)
|
||
|
||
def test_legacy_import_list_aliases_kept(self) -> None:
|
||
parser = self._parser()
|
||
args = parser.parse_args(["import", "--projects-json", "x.json"])
|
||
self.assertEqual(args.projects_json, "x.json")
|
||
args = parser.parse_args(["list", "--kind", "expenses", "--project-no", "X"])
|
||
self.assertEqual(args.kind, "expenses")
|
||
|
||
def test_render_html_unresolved_placeholder_falls_back(self) -> None:
|
||
parser = self._parser()
|
||
args = parser.parse_args(["render-html", "--report-id", "{{id}}"])
|
||
# 未解析占位符不应让渲染崩溃：handler 内部按缺省（最近 10 份）处理
|
||
self.assertEqual(args.report_id, "{{id}}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|