# -*- 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()