437 lines
17 KiB
Python
437 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""票据合规预审规则引擎金标准用例（离线，无网络无浏览器）。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date, timedelta
|
|
|
|
from _support import IsolatedDataRoot
|
|
|
|
_PASS = "pass"
|
|
_WARN = "warn"
|
|
_REJECT = "reject"
|
|
|
|
|
|
def _levels(items):
|
|
return {i["rule_id"]: i["level"] for i in items}
|
|
|
|
|
|
def _verdict_of(result):
|
|
return {r["invoice_number"]: r["verdict"] for r in result["reports"]}
|
|
|
|
|
|
class PrereviewRuleTestBase(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 _weekday(self, offset: int = 0) -> str:
|
|
"""返回一个周一~周四且不在节假日种子的日期（避免 R3 误触发）；从今天往回找，保证不是未来日期。"""
|
|
from db.connection import get_conn
|
|
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT holiday_date FROM holiday_calendar")
|
|
holidays = {str(r[0]) for r in cur.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
base = date.today() - timedelta(days=max(offset, 0))
|
|
for delta in range(0, 30):
|
|
d = base - timedelta(days=delta)
|
|
if d.weekday() <= 3 and d.isoformat() not in holidays:
|
|
return d.isoformat()
|
|
raise AssertionError("30 天内找不到工作日测试日期")
|
|
|
|
def _saturday(self) -> str:
|
|
base = date.today()
|
|
for delta in range(0, 14):
|
|
d = base + timedelta(days=delta)
|
|
if d.weekday() == 5:
|
|
return d.isoformat()
|
|
raise AssertionError("两周内找不到周六")
|
|
|
|
def _run(self, invoices, **context):
|
|
from service.prereview_service import run_prereview
|
|
|
|
ctx = {
|
|
"purpose": "",
|
|
"project_no": "",
|
|
"trip_start": "",
|
|
"trip_end": "",
|
|
"trip_city": "",
|
|
"staff_title": "",
|
|
}
|
|
ctx.update(context)
|
|
return run_prereview(invoices, ctx)
|
|
|
|
|
|
class TestCnyAmount(PrereviewRuleTestBase):
|
|
def test_parse_common_upper_amounts(self) -> None:
|
|
from util.cny_amount import parse_cny_upper
|
|
|
|
cases = {
|
|
"壹仟陆佰伍拾元整": 1650.0,
|
|
"壹佰贰拾叁元肆角伍分": 123.45,
|
|
"拾元整": 10.0,
|
|
"壹万贰仟零壹佰元整": 12100.0,
|
|
"人民币叁元整": 3.0,
|
|
}
|
|
for text, expected in cases.items():
|
|
with self.subTest(text=text):
|
|
self.assertEqual(parse_cny_upper(text), expected)
|
|
|
|
def test_parse_garbage_returns_none(self) -> None:
|
|
from util.cny_amount import parse_cny_upper
|
|
|
|
self.assertIsNone(parse_cny_upper(""))
|
|
self.assertIsNone(parse_cny_upper("abc"))
|
|
self.assertIsNone(parse_cny_upper("123.45"))
|
|
|
|
|
|
class TestFieldCompleteness(PrereviewRuleTestBase):
|
|
def test_f1_missing_fields_warn(self) -> None:
|
|
result = self._run([{"invoiceNumber": "24312000000012345678"}])
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["F1"], _WARN)
|
|
self.assertEqual(result["reports"][0]["verdict"], "待确认")
|
|
|
|
def test_f2_unknown_length_warn(self) -> None:
|
|
result = self._run([{"invoiceNumber": "12345", "issueDate": self._weekday(), "amount": 10, "sellerName": "x"}])
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["F2"], _WARN)
|
|
|
|
def test_f2_dianpiao_20_digits_pass(self) -> None:
|
|
result = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x"}])
|
|
items = _levels(result["reports"][0]["items"])
|
|
# 号码位数合法时不额外出 F2 项（无异常即通过）
|
|
self.assertEqual(items.get("F2", _PASS), _PASS)
|
|
|
|
def test_f3_future_date_reject(self) -> None:
|
|
future = (date.today() + timedelta(days=400)).isoformat()
|
|
result = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": future, "amount": 10, "sellerName": "x"}])
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["F3"], _REJECT)
|
|
self.assertEqual(result["reports"][0]["verdict"], "驳回")
|
|
|
|
def test_f4_words_mismatch_reject(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 1650.0,
|
|
"amountInWords": "壹佰元整",
|
|
"sellerName": "x",
|
|
}
|
|
]
|
|
)
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["F4"], _REJECT)
|
|
|
|
def test_f4_words_match_pass(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 1650.0,
|
|
"amountInWords": "壹仟陆佰伍拾元整",
|
|
"sellerName": "x",
|
|
}
|
|
]
|
|
)
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["F4"], _PASS)
|
|
|
|
|
|
class TestCrossVerification(PrereviewRuleTestBase):
|
|
def test_v1_unverified_warn(self) -> None:
|
|
result = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x"}])
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["V1"], _WARN)
|
|
|
|
def test_v1_manual_fail_rejects(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 10,
|
|
"sellerName": "x",
|
|
"manualVerifyResult": "fail",
|
|
}
|
|
]
|
|
)
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["V1"], _REJECT)
|
|
self.assertEqual(result["reports"][0]["verdict"], "驳回")
|
|
|
|
def test_v2_whitelist_pass_and_reject(self) -> None:
|
|
good = "91440101MA9XXYYY0A"
|
|
ok = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x", "buyerTaxNo": good}])
|
|
self.assertEqual(_levels(ok["reports"][0]["items"])["V2"], _PASS)
|
|
|
|
bad = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x", "buyerTaxNo": "00000000000000000X"}])
|
|
self.assertEqual(_levels(bad["reports"][0]["items"])["V2"], _REJECT)
|
|
|
|
def test_v3_outside_trip_window_warn(self) -> None:
|
|
from db.connection import get_conn
|
|
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT holiday_date FROM holiday_calendar")
|
|
holidays = {str(r[0]) for r in cur.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
trip_start = date.today() + timedelta(days=10)
|
|
trip_start = trip_start + timedelta(days=(7 - trip_start.weekday()) % 7) # 移到周一
|
|
issue = trip_start - timedelta(days=30)
|
|
for _ in range(10):
|
|
if issue.weekday() <= 3 and issue.isoformat() not in holidays:
|
|
break
|
|
issue = issue - timedelta(days=1)
|
|
result = self._run(
|
|
[{"invoiceNumber": "24312000000012345678", "issueDate": issue.isoformat(), "amount": 10, "sellerName": "x"}],
|
|
trip_start=trip_start.isoformat(),
|
|
trip_end=(trip_start + timedelta(days=2)).isoformat(),
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["V3"], _WARN)
|
|
|
|
def test_v3_inside_trip_window_pass(self) -> None:
|
|
start = self._weekday()
|
|
end_date = date.fromisoformat(start) + timedelta(days=2)
|
|
issue = date.fromisoformat(start) + timedelta(days=1)
|
|
result = self._run(
|
|
[{"invoiceNumber": "24312000000012345678", "issueDate": issue.isoformat(), "amount": 10, "sellerName": "x"}],
|
|
trip_start=start,
|
|
trip_end=end_date.isoformat(),
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["V3"], _PASS)
|
|
|
|
|
|
class TestExpenseStandards(PrereviewRuleTestBase):
|
|
def test_s1_over_lodging_cap_rejects(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 2400.0,
|
|
"sellerName": "酒店",
|
|
"expenseCategory": "住宿费",
|
|
"city": "北京",
|
|
"days": 2,
|
|
}
|
|
],
|
|
staff_title="副高及以上",
|
|
)
|
|
items = _levels(result["reports"][0]["items"])
|
|
self.assertEqual(items["S1"], _REJECT) # 1200 元/天 > 550
|
|
self.assertIn("550", result["reports"][0]["items"][_level_index(result, "S1")]["detail"])
|
|
|
|
def test_s1_within_cap_passes(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 1080.0,
|
|
"sellerName": "酒店",
|
|
"expenseCategory": "住宿费",
|
|
"city": "北京",
|
|
"days": 2,
|
|
}
|
|
],
|
|
staff_title="副高及以上",
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["S1"], _PASS) # 540 元/天 ≤ 550
|
|
|
|
def test_s1_unknown_city_warns(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 900.0,
|
|
"sellerName": "酒店",
|
|
"expenseCategory": "住宿费",
|
|
"city": "未知城镇",
|
|
"days": 1,
|
|
}
|
|
],
|
|
staff_title="副高及以上",
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["S1"], _WARN)
|
|
|
|
def test_s2_ticket_class_above_title_warns(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 100.0,
|
|
"sellerName": "铁路",
|
|
"expenseCategory": "交通费",
|
|
"ticketClass": "商务座",
|
|
}
|
|
],
|
|
staff_title="中级及以下",
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["S2"], _WARN)
|
|
|
|
def test_s3_project_hint_pass(self) -> None:
|
|
result = self._run(
|
|
[{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x"}],
|
|
project_no="KYGX2026001",
|
|
)
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["S3"], _PASS)
|
|
|
|
|
|
class TestRiskFlags(PrereviewRuleTestBase):
|
|
def test_r1_consecutive_numbers_warn(self) -> None:
|
|
issue = self._weekday()
|
|
result = self._run(
|
|
[
|
|
{"invoiceNumber": "24312000000012345678", "issueDate": issue, "amount": 10, "sellerName": "a"},
|
|
{"invoiceNumber": "24312000000012345679", "issueDate": issue, "amount": 10, "sellerName": "b"},
|
|
]
|
|
)
|
|
for report in result["reports"]:
|
|
self.assertEqual(_levels(report["items"])["R1"], _WARN)
|
|
|
|
def test_r2_batch_concentration_warn(self) -> None:
|
|
issue = self._weekday()
|
|
result = self._run(
|
|
[
|
|
{"invoiceNumber": f"243120000000123450{i:02d}", "issueDate": issue, "amount": 10, "sellerName": "同一商家", "sellerTaxNo": "91110000TEST00001X"}
|
|
for i in range(1, 4)
|
|
]
|
|
)
|
|
for report in result["reports"]:
|
|
self.assertEqual(_levels(report["items"])["R2"], _WARN)
|
|
|
|
def test_r3_weekend_invoicing_warns(self) -> None:
|
|
result = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._saturday(), "amount": 10, "sellerName": "x"}])
|
|
self.assertEqual(_levels(result["reports"][0]["items"])["R3"], _WARN)
|
|
|
|
|
|
class TestVerdictAggregation(PrereviewRuleTestBase):
|
|
def test_clean_invoice_passes(self) -> None:
|
|
result = self._run(
|
|
[
|
|
{
|
|
"invoiceNumber": "24312000000012345678",
|
|
"issueDate": self._weekday(),
|
|
"amount": 10.0,
|
|
"amountInWords": "拾元整",
|
|
"sellerName": "商店",
|
|
"manualVerifyResult": "pass",
|
|
"buyerTaxNo": "91440101MA9XXYYY0A",
|
|
}
|
|
]
|
|
)
|
|
self.assertEqual(result["reports"][0]["verdict"], "通过")
|
|
self.assertEqual(result["verdict_counts"], {"通过": 1, "待确认": 0, "驳回": 0})
|
|
|
|
def test_batch_counts_and_batch_no_shared(self) -> None:
|
|
issue = self._weekday()
|
|
result = self._run(
|
|
[
|
|
{"invoiceNumber": "24312000000012345678", "issueDate": issue, "amount": 10, "sellerName": "a", "manualVerifyResult": "pass", "buyerTaxNo": "91440101MA9XXYYY0A"},
|
|
{"invoiceNumber": "24312000000012345999", "issueDate": issue, "amount": 10, "sellerName": "b"},
|
|
]
|
|
)
|
|
self.assertEqual(result["count"], 2)
|
|
self.assertEqual(result["verdict_counts"], {"通过": 1, "待确认": 1, "驳回": 0})
|
|
batch_nos = {r["batch_no"] for r in result["reports"]}
|
|
self.assertEqual(len(batch_nos), 1)
|
|
self.assertEqual(len(result["batch_no"]), 16)
|
|
|
|
def test_report_file_written(self) -> None:
|
|
result = self._run([{"invoiceNumber": "24312000000012345678", "issueDate": self._weekday(), "amount": 10, "sellerName": "x"}])
|
|
path = result["reports"][0]["report_file"]
|
|
self.assertTrue(path.endswith(".html"))
|
|
with open(path, encoding="utf-8") as f:
|
|
html_text = f.read()
|
|
self.assertIn("初审意见单", html_text)
|
|
self.assertIn("不构成最终报销结论", html_text)
|
|
|
|
def test_empty_input_raises(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
self._run([])
|
|
|
|
|
|
def _level_index(result, rule_id: str) -> int:
|
|
items = result["reports"][0]["items"]
|
|
for idx, item in enumerate(items):
|
|
if item["rule_id"] == rule_id:
|
|
return idx
|
|
raise AssertionError(f"rule {rule_id} not found")
|
|
|
|
|
|
class TestMarkVerified(unittest.TestCase):
|
|
def test_fail_upgrades_verdict_to_reject(self) -> None:
|
|
with IsolatedDataRoot():
|
|
from db.connection import init_db
|
|
from service.prereview_service import run_prereview
|
|
from service.standards_service import get_report, mark_verified
|
|
|
|
init_db()
|
|
result = run_prereview(
|
|
[{"invoiceNumber": "24312000000012345678", "issueDate": "2099-01-01", "amount": 10, "sellerName": "x"}],
|
|
{"purpose": "", "project_no": "", "trip_start": "", "trip_end": "", "trip_city": "", "staff_title": ""},
|
|
)
|
|
report_id = result["reports"][0]["report_id"]
|
|
payload = mark_verified(report_id, "fail", "查验不符")
|
|
self.assertEqual(payload["verify_status"], "已查验不符")
|
|
report = get_report(report_id)
|
|
self.assertEqual(report["verdict"], "驳回")
|
|
self.assertEqual(report["reject_json"][-1]["rule_id"], "V1")
|
|
|
|
|
|
class TestWhitelistAndStandardsServices(unittest.TestCase):
|
|
def test_whitelist_add_duplicate_remove(self) -> None:
|
|
with IsolatedDataRoot():
|
|
from db.connection import init_db
|
|
from service.standards_service import add_whitelist, list_whitelist, remove_whitelist
|
|
|
|
init_db()
|
|
add_whitelist("91440101TESTAAAA01", "测试单位", "n")
|
|
with self.assertRaises(ValueError):
|
|
add_whitelist("91440101TESTAAAA01", "重复税号")
|
|
self.assertEqual(len(list_whitelist()), 2) # 种子 1 条 + 新增 1 条
|
|
remove_whitelist("91440101TESTAAAA01")
|
|
with self.assertRaises(ValueError):
|
|
remove_whitelist("91440101TESTAAAA01")
|
|
|
|
def test_set_standard_upsert(self) -> None:
|
|
with IsolatedDataRoot():
|
|
from db.connection import init_db
|
|
from service.standards_service import list_standards, upsert_standard
|
|
|
|
init_db()
|
|
payload = upsert_standard(
|
|
category="住宿费",
|
|
city_tier="一线城市",
|
|
staff_title="副高及以上",
|
|
max_amount=600.0,
|
|
unit="元/人/天",
|
|
basis_doc="《差旅费管理办法》",
|
|
clause="第十条",
|
|
)
|
|
self.assertEqual(payload["action"], "updated")
|
|
row = next(s for s in list_standards("住宿费") if s["city_tier"] == "一线城市" and s["staff_title"] == "副高及以上")
|
|
self.assertEqual(row["max_amount"], 600.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|