"""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()