f90f742c1e
All checks were successful
技能自动化发布 / release (push) Successful in 9s
- 状态标签：C3 触发且 D 未判 → 待补录，优先级对齐需求 §6.2 - broadcast-monthly 逐项目增量 emit，任务中心可见进行中进度 - progress_reports 回填 project_id；新增 items_json/updated_at；旧库重建式安全迁移（含回归测试） - 简报：TITLE 转义防注入、章节对齐 §10（独立「暂付款/结题提醒」）、页脚逐字统一 - 金标准测试 +17：±10%/20% 阈值恰界、C2 钳位、N2 反例、C3 逐字段、HTML 转义、async+cron 定位守护、迁移回归 - 文档 v1.1（开发文档 + REQUIREMENTS）：ERROR:<CODE> <message>、reports_brief、--all-status、skill-detail 表外挂载、C1 level=warn - 落盘验收报告 acceptance-reports/20260924-105401-acceptance.md（338 测试全绿） - 版本 1.0.1 → 1.0.2（constants / SKILL.md / CHANGELOG 三处对齐）
399 lines
16 KiB
Python
399 lines
16 KiB
Python
# -*- 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("正常"))
|
||
|
||
def test_on_progress_called_per_project(self) -> None:
|
||
from service.fund_service import run_broadcast
|
||
|
||
seen: list[tuple[int, int, str]] = []
|
||
result = run_broadcast(
|
||
[_project(), _project(projectNo="TEST0002")],
|
||
AS_OF,
|
||
on_progress=lambda i, t, r: seen.append((i, t, r["project_no"])),
|
||
)
|
||
self.assertEqual(seen, [(1, 2, "TEST0001"), (2, 2, "TEST0002")])
|
||
self.assertEqual(result["count"], 2)
|
||
|
||
def test_report_links_project_id_and_items_json(self) -> None:
|
||
import json
|
||
|
||
from db.connection import get_conn
|
||
from service.fund_service import query_projects, run_broadcast
|
||
from service.project_service import upsert_project
|
||
|
||
upsert_project({
|
||
"project_no": "TEST0001",
|
||
"total_received": 300000.0,
|
||
"total_spent": 60000.0,
|
||
"start_date": "2026-01-01",
|
||
"end_date": "2026-12-31",
|
||
"frozen_amount": 15000.0,
|
||
"frozen_since": "2026-04-10",
|
||
})
|
||
projects = query_projects(include_all_status=True)
|
||
self.assertEqual(len(projects), 1)
|
||
run_broadcast(projects, AS_OF)
|
||
conn = get_conn()
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT project_id, items_json FROM progress_reports WHERE project_no = ?",
|
||
("TEST0001",),
|
||
).fetchone()
|
||
finally:
|
||
conn.close()
|
||
self.assertEqual(row[0], projects[0]["id"])
|
||
items = json.loads(row[1])
|
||
self.assertTrue(any(i["rule_id"] == "D1" for i in items))
|
||
self.assertTrue(any(i["rule_id"] == "N1" for i in items))
|
||
|
||
def test_report_escapes_html_injection(self) -> None:
|
||
result = self._run([_project(projectName="<script>alert(1)</script>")])
|
||
with open(result["reports"][0]["report_file"], encoding="utf-8") as f:
|
||
html_text = f.read()
|
||
self.assertNotIn("<script>alert(1)</script>", html_text)
|
||
self.assertIn("<script>", html_text)
|
||
|
||
def test_report_contains_status_lights_and_reminders(self) -> None:
|
||
result = self._run([
|
||
_project(projectNo="LAG001", frozenAmount=20000.0, frozenSince="2026-01-20"),
|
||
_project(projectNo="FAST001", totalReceived=400000.0, totalSpent=380000.0),
|
||
_project(projectNo="OK001", totalReceived=400000.0, totalSpent=350000.0),
|
||
_project(projectNo="CLO001", endDate="2026-10-15", totalReceived=100000.0, totalSpent=5000.0),
|
||
])
|
||
by_no = {r["project_no"]: r for r in result["reports"]}
|
||
|
||
def _read(project_no: str) -> str:
|
||
with open(by_no[project_no]["report_file"], encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
lag_html = _read("LAG001")
|
||
self.assertIn("🔴", lag_html)
|
||
self.assertIn("冲销", lag_html)
|
||
self.assertIn("暂付款 / 结题提醒", lag_html)
|
||
self.assertIn("🟢", _read("OK001"))
|
||
self.assertIn("🟠", _read("FAST001"))
|
||
clo_html = _read("CLO001")
|
||
self.assertIn("距执行期结束", clo_html)
|
||
self.assertIn("暂付款 / 结题提醒", clo_html)
|
||
|
||
|
||
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"])
|
||
|
||
|
||
class TestThresholdBoundaries(unittest.TestCase):
|
||
"""验收 3：恰在 −10% / +20% 阈值边界的归类（默认阈值 0.10 / 0.20）。"""
|
||
|
||
def test_d1_exactly_minus_10_percent_is_normal(self) -> None:
|
||
result = _compute(totalReceived=100000.0, totalSpent=60050.0)
|
||
self.assertEqual(result["time_progress"], 0.7005)
|
||
self.assertEqual(result["deviation"], -0.1)
|
||
self.assertEqual(result["status_tag"], "正常") # 严格小于才判滞后
|
||
|
||
def test_d1_just_below_minus_10_percent_is_lag(self) -> None:
|
||
result = _compute(totalReceived=100000.0, totalSpent=60040.0) # dev=-0.1001
|
||
self.assertEqual(result["status_tag"], "严重滞后")
|
||
|
||
def test_d2_exactly_plus_20_percent_is_normal(self) -> None:
|
||
result = _compute(totalReceived=100000.0, totalSpent=90050.0)
|
||
self.assertEqual(result["deviation"], 0.2)
|
||
self.assertEqual(result["status_tag"], "正常")
|
||
|
||
def test_d2_just_above_plus_20_percent_is_overshoot(self) -> None:
|
||
result = _compute(totalReceived=100000.0, totalSpent=90060.0) # dev=+0.2001
|
||
self.assertEqual(result["status_tag"], "支出异常偏快")
|
||
|
||
|
||
class TestC2Clamp(unittest.TestCase):
|
||
def test_before_start_clamps_to_zero(self) -> None:
|
||
result = _compute(startDate="2026-10-01", endDate="2026-12-31")
|
||
self.assertEqual(result["time_progress"], 0.0)
|
||
|
||
def test_after_end_clamps_to_one(self) -> None:
|
||
result = _compute(startDate="2026-01-01", endDate="2026-08-31")
|
||
self.assertEqual(result["time_progress"], 1.0)
|
||
|
||
|
||
class TestC3StatusPriority(unittest.TestCase):
|
||
"""状态标签优先级：严重滞后＞偏快＞正常（D3 已判）＞待补录（C3）＞未到账（C1）。"""
|
||
|
||
def test_missing_end_date_marks_incomplete(self) -> None:
|
||
result = _compute(endDate=None)
|
||
self.assertIsNotNone(result["spend_rate"])
|
||
self.assertEqual(result["status_tag"], "待补录")
|
||
|
||
def test_missing_start_date_marks_incomplete(self) -> None:
|
||
result = _compute(startDate=None)
|
||
self.assertEqual(result["status_tag"], "待补录")
|
||
|
||
def test_missing_total_received_marks_incomplete(self) -> None:
|
||
result = _compute(totalReceived=None)
|
||
self.assertEqual(result["status_tag"], "待补录")
|
||
|
||
def test_missing_total_approved_keeps_normal_when_d3_judged(self) -> None:
|
||
result = _compute(totalApproved=None, totalReceived=400000.0, totalSpent=350000.0)
|
||
self.assertEqual(result["status_tag"], "正常") # 正常＞待补录
|
||
c3 = next(i for i in result["items"] if i["rule_id"] == "C3")
|
||
self.assertIn("总立项经费", c3["detail"])
|
||
|
||
|
||
class TestN2Negative(unittest.TestCase):
|
||
def test_closing_soon_but_rate_at_least_80pct_no_reminder(self) -> None:
|
||
result = _compute(endDate="2026-10-15", totalReceived=100000.0, totalSpent=80000.0)
|
||
self.assertFalse(any(i["rule_id"] == "N2" for i in result["items"]))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|