194 lines
9.7 KiB
Python
194 lines
9.7 KiB
Python
 
|
"""T05-a 历史时间线测试：AC-02 与 BR-02 的机械断言。
|
|||
|
|
|
|||
|
|
核心是两条"不许编"：
|
|||
|
|
- 不许编造发生时间（缺日期 → None + 缺证提示，导入时间不得冒充）
|
|||
|
|
- 不许编造经过（只有最终报告 → 0 事件 + 过程不完整，不生成虚构问询顺序）
|
|||
|
|
|
|||
|
|
用临时库；不触碰真实数据根；不做模型调用。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import tempfile
|
|||
|
|
import unittest
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
from contracts.prescreen import CaseTimeline
|
|||
|
|
from services import assistant_store as store
|
|||
|
|
from services import history_cases
|
|||
|
|
from services.assistant_materials import material
|
|||
|
|
|
|||
|
|
HEADER = "date,event_type,question,judgment_before,judgment_after\n"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TimelineTestCase(unittest.TestCase):
|
|||
|
|
def setUp(self):
|
|||
|
|
self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|||
|
|
self.db_path = Path(self._tmp.name) / "history.sqlite3"
|
|||
|
|
store.initialize(self.db_path)
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
for project_id in ("DEMO-A", "DEMO-B"):
|
|||
|
|
store.put(db, "project", {
|
|||
|
|
"id": project_id, "data_domain": "demo", "revision": 1,
|
|||
|
|
"facts": {}, "values": {}, "sources": {}, "selected_sources": {},
|
|||
|
|
"document_ids": [], "pending_candidates": [], "data_revision": 1,
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def tearDown(self):
|
|||
|
|
self._tmp.cleanup()
|
|||
|
|
|
|||
|
|
def add_document(self, content: bytes, name: str, project_id: str = "DEMO-A"):
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
document = material(project_id, name, content, "tester")
|
|||
|
|
store.put(db, "document", document, immutable=True)
|
|||
|
|
project = store.get(db, "project", project_id)
|
|||
|
|
project["document_ids"].append(document["id"])
|
|||
|
|
store.put(db, "project", project)
|
|||
|
|
return document
|
|||
|
|
|
|||
|
|
def build(self, project_id: str = "DEMO-A"):
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
return history_cases.build(db, project_id, "demo")
|
|||
|
|
|
|||
|
|
def load(self, project_id: str = "DEMO-A"):
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
return history_cases.load(db, project_id)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class NoFabricationTests(TimelineTestCase):
|
|||
|
|
def test_final_report_only_yields_no_events_and_says_incomplete(self):
|
|||
|
|
# 14 §4：只有最终报告时标"过程不完整"，不生成虚构经过
|
|||
|
|
self.add_document("尽调结论：该客户经营稳定，建议通过。\n".encode("utf-8-sig"), "最终报告.txt")
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertEqual(timeline["events"], [], "不得凭最终报告编造时间线")
|
|||
|
|
self.assertEqual(timeline["status"], "partial")
|
|||
|
|
self.assertTrue(any("过程材料不完整" in w for w in timeline["warnings"]))
|
|||
|
|
|
|||
|
|
def test_no_documents_at_all_still_reports_incomplete(self):
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertEqual(timeline["events"], [])
|
|||
|
|
self.assertEqual(timeline["status"], "partial")
|
|||
|
|
self.assertTrue(any("过程材料不完整" in w for w in timeline["warnings"]))
|
|||
|
|
|
|||
|
|
def test_unknown_columns_are_not_guessed(self):
|
|||
|
|
# 列名不认识就忽略，不猜哪一列是问题/日期
|
|||
|
|
self.add_document("姓名,金额,备注\n张三,100,随便写点什么\n".encode("utf-8-sig"), "无关表.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertEqual(timeline["events"], [])
|
|||
|
|
|
|||
|
|
def test_missing_date_is_null_not_the_import_time(self):
|
|||
|
|
self.add_document((HEADER + ",问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
event = timeline["events"][0]
|
|||
|
|
self.assertIsNone(event["occurred_at"], "缺日期必须是 None，不能用导入时间顶替")
|
|||
|
|
self.assertIsNotNone(event["recorded_at"], "录入时间必填，但它不表示发生时间")
|
|||
|
|
self.assertNotEqual(event["occurred_at"], event["recorded_at"])
|
|||
|
|
|
|||
|
|
def test_unparseable_date_is_not_coerced(self):
|
|||
|
|
self.add_document((HEADER + "去年底,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertIsNone(timeline["events"][0]["occurred_at"])
|
|||
|
|
self.assertTrue(any("未标注发生时间" in w for w in timeline["warnings"]))
|
|||
|
|
|
|||
|
|
def test_placeholder_values_are_treated_as_missing(self):
|
|||
|
|
for placeholder in ("缺失", "未知", "-", "null", "无"):
|
|||
|
|
with self.subTest(value=placeholder):
|
|||
|
|
self.add_document((HEADER + f"{placeholder},问询,请说明回款来源,,\n").encode("utf-8-sig"),
|
|||
|
|
f"历史{placeholder}.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
undated = [e for e in timeline["events"] if e["occurred_at"] is None]
|
|||
|
|
self.assertTrue(undated, f"{placeholder} 应被视为缺日期")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ExtractionTests(TimelineTestCase):
|
|||
|
|
def test_dated_rows_become_ready(self):
|
|||
|
|
rows = HEADER + "2024-03-11,问询,请说明回款来源,材料不足,已补充\n2024-04-02,决定,评审会通过,,\n"
|
|||
|
|
self.add_document(rows.encode("utf-8-sig"), "历史.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertEqual(timeline["status"], "ready")
|
|||
|
|
self.assertEqual([e["question"] for e in timeline["events"]],
|
|||
|
|
["请说明回款来源", "评审会通过"])
|
|||
|
|
self.assertTrue(all(e["occurred_at"] for e in timeline["events"]))
|
|||
|
|
|
|||
|
|
def test_one_undated_event_makes_the_whole_timeline_partial(self):
|
|||
|
|
rows = HEADER + "2024-03-11,问询,请说明回款来源,,\n,决定,评审会通过,,\n"
|
|||
|
|
self.add_document(rows.encode("utf-8-sig"), "历史.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
self.assertEqual(timeline["status"], "partial")
|
|||
|
|
self.assertEqual(len(timeline["events"]), 2)
|
|||
|
|
|
|||
|
|
def test_events_carry_evidence_ids_for_traceability(self):
|
|||
|
|
self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
timeline = self.build()
|
|||
|
|
event = timeline["events"][0]
|
|||
|
|
self.assertEqual(len(event["evidence_ids"]), 1)
|
|||
|
|
# 证据必须能在文档的 chunk 里定位到（AC-02：能定位原文）
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
document = store.get(db, "document", store.listing(db, "document", "DEMO-A")[0]["id"])
|
|||
|
|
ids = {chunk["evidence_id"] for chunk in document["chunks"]}
|
|||
|
|
self.assertIn(event["evidence_ids"][0], ids)
|
|||
|
|
|
|||
|
|
def test_event_type_aliases_and_conservative_default(self):
|
|||
|
|
rows = (HEADER
|
|||
|
|
+ "2024-01-01,问询,问题一,,\n"
|
|||
|
|
+ "2024-01-02,决定,结论一,,\n"
|
|||
|
|
+ "2024-01-03,完全没听过的类型,问题三,,\n")
|
|||
|
|
self.add_document(rows.encode("utf-8-sig"), "历史.csv")
|
|||
|
|
types = [e["event_type"] for e in self.build()["events"]]
|
|||
|
|
self.assertEqual(types, ["question", "decision", "question"])
|
|||
|
|
|
|||
|
|
def test_conditions_are_split(self):
|
|||
|
|
rows = "date,question,conditions\n2024-01-01,请补充说明,需担保；限三个月\n"
|
|||
|
|
self.add_document(rows.encode("utf-8-sig"), "历史.csv")
|
|||
|
|
self.assertEqual(self.build()["events"][0]["conditions"], ["需担保", "限三个月"])
|
|||
|
|
|
|||
|
|
def test_chinese_date_format_is_accepted(self):
|
|||
|
|
self.add_document((HEADER + "2024年3月11日,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
self.assertEqual(self.build()["events"][0]["occurred_at"][:10], "2024-03-11")
|
|||
|
|
|
|||
|
|
def test_undated_events_are_sorted_last_not_shuffled_into_order(self):
|
|||
|
|
rows = HEADER + ",问询,无日期的问题,,\n2024-03-11,问询,有日期的问题,,\n"
|
|||
|
|
self.add_document(rows.encode("utf-8-sig"), "历史.csv")
|
|||
|
|
events = self.build()["events"]
|
|||
|
|
self.assertEqual(events[0]["question"], "有日期的问题")
|
|||
|
|
self.assertIsNone(events[1]["occurred_at"])
|
|||
|
|
self.assertEqual([e["question"] for e in history_cases.undated_events({"events": events})],
|
|||
|
|
["无日期的问题"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
class IsolationAndRevisionTests(TimelineTestCase):
|
|||
|
|
def test_other_projects_documents_are_not_included(self):
|
|||
|
|
self.add_document((HEADER + "2024-03-11,问询,DEMO-A 的问题,,\n").encode("utf-8-sig"), "a.csv", "DEMO-A")
|
|||
|
|
self.add_document((HEADER + "2024-05-05,问询,DEMO-B 的问题,,\n").encode("utf-8-sig"), "b.csv", "DEMO-B")
|
|||
|
|
timeline = self.build("DEMO-A")
|
|||
|
|
self.assertEqual([e["question"] for e in timeline["events"]], ["DEMO-A 的问题"])
|
|||
|
|
|
|||
|
|
def test_rebuild_bumps_revision_on_the_same_object(self):
|
|||
|
|
self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
first = self.build()
|
|||
|
|
second = self.build()
|
|||
|
|
self.assertEqual(first["id"], second["id"], "重新提炼应是同一对象的新 revision")
|
|||
|
|
self.assertEqual(first["revision"], 1)
|
|||
|
|
self.assertEqual(second["revision"], 2)
|
|||
|
|
|
|||
|
|
def test_stored_timeline_passes_the_contract(self):
|
|||
|
|
self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv")
|
|||
|
|
self.build()
|
|||
|
|
CaseTimeline(**self.load())
|
|||
|
|
|
|||
|
|
def test_load_rejects_a_foreign_project_lookup(self):
|
|||
|
|
self.build("DEMO-A")
|
|||
|
|
with store.transaction(self.db_path) as db:
|
|||
|
|
with self.assertRaises(KeyError):
|
|||
|
|
history_cases.load(db, "DEMO-B")
|
|||
|
|
|
|||
|
|
def test_ready_is_impossible_with_undated_events(self):
|
|||
|
|
# 契约层兜底：即便调用方算错状态，落库前的校验也会拦住
|
|||
|
|
with self.assertRaises(Exception):
|
|||
|
|
CaseTimeline(id="t", project_id="DEMO-A", data_domain="demo", revision=1,
|
|||
|
|
status="ready", created_at=store.now(),
|
|||
|
|
events=[{"id": "e1", "event_type": "question",
|
|||
|
|
"recorded_at": store.now(), "question": "q"}])
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
unittest.main()
|