# -*- coding: utf-8 -*- """经费播报计算内核金标准用例（离线、确定性，as_of 固定日期）。""" from __future__ import annotations import unittest from datetime import date from _support import IsolatedDataRoot AS_OF = date(2026, 9, 13) # 基准日：2026-01-01 起止 2026-12-31 的项目序时进度 ≈ 0.7005 def _project(**overrides) -> dict: base = { "projectNo": "TEST0001", "projectName": "测试项目", "projectType": "纵向", "piName": "张三", "department": "化材学院", "totalApproved": 600000.0, "totalReceived": 300000.0, "totalSpent": 60000.0, "frozenAmount": 0.0, "startDate": "2026-01-01", "endDate": "2026-12-31", "status": "在研", } base.update(overrides) return base def _compute(rules: dict | None = None, **overrides) -> dict: from service.fund_service import _DEFAULT_RULES, compute_project return compute_project(_project(**overrides), dict(_DEFAULT_RULES if rules is None else rules), AS_OF) class TestStep1Calculation(unittest.TestCase): def test_c2_time_progress_by_days(self) -> None: result = _compute() # 2026-01-01 ~ 2026-12-31 共 364 天，至 2026-09-13 过 255 天 self.assertAlmostEqual(result["time_progress"], 255 / 364, places=4) def test_c1_no_fund_marks_unknown(self) -> None: result = _compute(totalReceived=0, totalSpent=0) self.assertEqual(result["status_tag"], "未到账") self.assertIsNone(result["spend_rate"]) def test_c3_incomplete_ledger(self) -> None: result = _compute(totalSpent=None, endDate=None) self.assertEqual(result["status_tag"], "待补录") c3 = next(i for i in result["items"] if i["rule_id"] == "C3") self.assertIn("已报销金额", c3["detail"]) def test_spend_rate_rounded(self) -> None: result = _compute() self.assertEqual(result["spend_rate"], 0.2) class TestStep2Diagnosis(unittest.TestCase): def test_d1_lag(self) -> None: result = _compute() self.assertEqual(result["status_tag"], "严重滞后") d1 = next(i for i in result["items"] if i["rule_id"] == "D1") self.assertIn("低于序时进度", d1["detail"]) def test_d3_normal(self) -> None: result = _compute(totalReceived=400000.0, totalSpent=350000.0) self.assertEqual(result["status_tag"], "正常") # dev ≈ +0.175 def test_d2_overshoot(self) -> None: result = _compute(totalReceived=400000.0, totalSpent=380000.0) self.assertEqual(result["status_tag"], "支出异常偏快") # dev ≈ +0.2495 def test_threshold_boundary_not_lag_when_equal(self) -> None: # 偏差恰等于 -0.10 时不判滞后（严格小于） result = _compute(rules={"lag_threshold": 0.5005, "overshoot_threshold": 5.0}) self.assertEqual(result["status_tag"], "正常") def test_rules_override_changes_verdict(self) -> None: result = _compute(rules={"lag_threshold": 0.60, "overshoot_threshold": 0.20}) # dev ≈ -0.5005 未超过 -0.60 → 正常 self.assertEqual(result["status_tag"], "正常") class TestStep3Reminders(unittest.TestCase): def test_n1_pending_aging_overdue(self) -> None: result = _compute(frozenAmount=15000.0, frozenSince="2026-04-10") n1 = next(i for i in result["items"] if i["rule_id"] == "N1") self.assertEqual((AS_OF - date(2026, 4, 10)).days, 156) self.assertIn("156", n1["detail"]) self.assertIn("冲销", n1["suggestion"]) def test_n1_pending_within_window_silent(self) -> None: result = _compute(frozenAmount=15000.0, frozenSince="2026-08-01") self.assertFalse(any(i["rule_id"] == "N1" for i in result["items"])) def test_n1_pending_since_missing_asks_backfill(self) -> None: result = _compute(frozenAmount=15000.0) n1 = next(i for i in result["items"] if i["rule_id"] == "N1") self.assertIn("补录", n1["suggestion"]) def test_n2_closing_soon_warns(self) -> None: result = _compute(endDate="2026-10-15", totalReceived=100000.0, totalSpent=5000.0) n2 = next(i for i in result["items"] if i["rule_id"] == "N2") self.assertIn("32", n2["detail"]) def test_n2_period_over_asks_settlement(self) -> None: result = _compute(endDate="2026-08-31", totalReceived=100000.0, totalSpent=90000.0) n2 = next(i for i in result["items"] if i["rule_id"] == "N2") self.assertIn("结题决算", n2["name"]) class TestSummaryText(unittest.TestCase): def test_three_part_summary_no_double_period(self) -> None: result = _compute(frozenAmount=15000.0, frozenSince="2026-04-10") lines = result["summary"].split("\n") self.assertEqual(len(lines), 3) self.assertTrue(lines[0].startswith("您负责的《测试项目》")) self.assertIn("执行状态严重滞后", lines[0]) self.assertTrue(lines[1].startswith("诊断：")) self.assertTrue(lines[2].startswith("建议：")) self.assertNotIn("。。", result["summary"]) self.assertNotIn("；。", result["summary"]) def test_summary_numbers_from_ledger_only(self) -> None: result = _compute() self.assertIn("60.00 万元", result["summary"]) self.assertIn("20.0%", result["summary"]) class TestBroadcastKernel(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 _run(self, projects): from service.fund_service import run_broadcast return run_broadcast(projects, AS_OF) def test_broadcast_writes_reports_and_brief(self) -> None: result = self._run([ _project(), _project(projectNo="TEST0002", totalReceived=400000.0, totalSpent=350000.0), ]) self.assertEqual(result["count"], 2) self.assertEqual(result["status_counts"], {"严重滞后": 1, "正常": 1}) self.assertTrue(result["brief_file"].endswith(".html")) for report in result["reports"]: self.assertTrue(report["report_file"].endswith(".html")) with open(report["report_file"], encoding="utf-8") as f: html_text = f.read() self.assertIn("经费执行简报", html_text) self.assertIn("以财务系统账面为准", html_text) def test_empty_projects_raises(self) -> None: with self.assertRaises(ValueError): self._run([]) def test_monthly_brief_contains_status_rows(self) -> None: result = self._run([_project()]) with open(result["brief_file"], encoding="utf-8") as f: html_text = f.read() self.assertIn("月度", html_text) self.assertIn("bar-fill fill-spend", html_text) def test_status_ranking_in_summary(self) -> None: result = self._run([ _project(), _project(projectNo="TEST0002", totalReceived=400000.0, totalSpent=350000.0), ]) brief_html = open(result["brief_file"], encoding="utf-8").read() self.assertLess(brief_html.find("严重滞后"), brief_html.find("正常")) class TestProjectService(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 _upsert(self, **overrides): from service.project_service import upsert_project fields = {"project_no": "TEST0001", "total_received": 300000.0, "total_spent": 60000.0} fields.update(overrides) return upsert_project(fields) def test_upsert_create_then_partial_update(self) -> None: payload = self._upsert() self.assertEqual(payload["action"], "created") payload = self._upsert(total_spent=90000.0) self.assertEqual(payload["action"], "updated") from service.project_service import list_projects row = list_projects()[0] self.assertEqual(row["total_spent"], 90000.0) self.assertEqual(row["total_received"], 300000.0) # 未提供字段不覆盖 self.assertEqual(row["project_name"], None) # 从未提供 def test_upsert_validations(self) -> None: with self.assertRaises(ValueError): self._upsert(project_no="") with self.assertRaises(ValueError): self._upsert(total_received=-1) with self.assertRaises(ValueError): self._upsert(start_date="2026/13/40") with self.assertRaises(ValueError): self._upsert(start_date="2026-01-01", end_date="2025-01-01") with self.assertRaises(ValueError): self._upsert(status="草稿") def test_import_projects(self) -> None: from service.project_service import import_projects payload = import_projects([ _project(), _project(projectNo="TEST0002"), ]) self.assertEqual(payload["created"], 2) payload = import_projects([_project(totalSpent=80000.0)]) self.assertEqual(payload["updated"], 1) def test_set_rule_and_effect(self) -> None: from service.project_service import set_rule, view_rules payload = set_rule("lag_threshold", 0.60) self.assertEqual(payload["threshold_value"], 0.60) rules = {r["rule_key"]: r["threshold_value"] for r in view_rules()} self.assertEqual(rules["lag_threshold"], 0.60) with self.assertRaises(ValueError): set_rule("unknown_key", 0.5) with self.assertRaises(ValueError): set_rule("lag_threshold", -1) def test_stats(self) -> None: from service.project_service import import_projects, stats import_projects([_project()]) from service.fund_service import run_broadcast run_broadcast([_project()], AS_OF) payload = stats() self.assertEqual(payload["project_total"], 1) self.assertEqual(payload["reports_by_tag"].get("严重滞后"), 1) self.assertTrue(payload["lag_top"]) if __name__ == "__main__": unittest.main()