qiaoqichaoandClaude Code
04ef3972a8
T05 卡片要求"先 timeline，再候选，再编辑/测试/审核"。本批是第一步。 新增 contracts：DecisionEvent（区分 occurred_at 与 recorded_at）、CaseTimeline。 新增 services/history_cases.py：确定性提炼，无模型调用。 两条硬规则的实现： - **不编造发生时间**（BR-02）：occurred_at 只取材料里记载的日期；缺失/占位值 （"-"、"缺失"、"未知"）或无法解析时保持 None 并给缺证提示。recorded_at 必填， 但它**不表示发生时间**——有测试断言二者不相等 - **不编造经过**（14 §4）：只有最终报告或非结构化材料时产出 **0 个事件** 并标 `partial` + "过程材料不完整"，绝不生成虚构的问询顺序 其它：列名靠别名表识别，**未识别的列忽略而非猜测**；未声明事件类型时保守记为 question 而不丢弃该行；事件携带行级 evidence_id 以便回溯原文定位；同项目重复提炼 是同一对象 revision 递增（03 §4）；缺日期事件排序时置于末尾，不被打乱成"近期"。 实现中修掉一处自己的问题：build() 初版调了两次 store.put（首次建库 revision 会变 2）， 且校验发生在落库之后——非法时间线本可能入库。现改为先校验再单次 put。 测试：新增 test_history_cases.py 18 项（不编造 6 + 提炼 6 + 隔离与版本 6）， 含"缺日期为 None 且不等于导入时间""只有最终报告不得生成时间线"等反例。 全量 290 项 OK（272 → 290）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code <noreply@anthropic.com>
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()
|