feat(v2): T05-a 历史过程时间线（缺时间不编造）
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>
This commit is contained in:
Claude Code
@@ -366,3 +366,52 @@ class DatasetManifest(Contract):
|
||||
if self.synthetic and self.data_domain != DataDomain.demo:
|
||||
raise ValueError("合成数据必须为 demo 数据域")
|
||||
return self
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 历史过程（T05）
|
||||
class DecisionEvent(Contract):
|
||||
"""历史过程中的一次问询/证据/动作/意见/决定（docs/v2/03-detailed-design.md §3）。
|
||||
|
||||
时间语义严格区分（BR-02）：
|
||||
- `occurred_at` 是材料里**记载**的发生时间，缺失时保持 None，**不得**用录入时间填充
|
||||
- `recorded_at` 是录入时间，必填，但**不能**被当作发生时间
|
||||
"""
|
||||
|
||||
id: Identifier
|
||||
event_type: Literal["question", "evidence", "action", "opinion", "decision"]
|
||||
occurred_at: AwareDatetime | None = None
|
||||
recorded_at: AwareDatetime
|
||||
question: str = Field(default="", max_length=DESC_MAX)
|
||||
evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX)
|
||||
judgment_before: str | None = Field(default=None, max_length=DESC_MAX)
|
||||
judgment_after: str | None = Field(default=None, max_length=DESC_MAX)
|
||||
conditions: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_journal(self) -> DecisionEvent:
|
||||
# 事件本身必须有可读内容，否则只是空壳
|
||||
if not (self.question or self.judgment_before or self.judgment_after):
|
||||
raise ValueError("决策事件必须至少包含问题、判断前或判断后之一")
|
||||
return self
|
||||
|
||||
|
||||
class CaseTimeline(Contract):
|
||||
"""历史过程时间线。`partial` 表示过程材料不完整（如只有最终报告）。"""
|
||||
|
||||
id: Identifier
|
||||
project_id: Identifier
|
||||
data_domain: DataDomain
|
||||
revision: int = Field(ge=1)
|
||||
status: Literal["processing", "ready", "partial", "failed"]
|
||||
created_at: AwareDatetime
|
||||
events: list[DecisionEvent] = Field(default_factory=list, max_length=500)
|
||||
warnings: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_integrity(self) -> CaseTimeline:
|
||||
ids = [event.id for event in self.events]
|
||||
if len(set(ids)) != len(ids):
|
||||
raise ValueError("同一时间线内事件 id 必须唯一")
|
||||
if self.status == "ready" and any(e.occurred_at is None for e in self.events):
|
||||
raise ValueError("存在缺发生时间的事件时状态不得为 ready")
|
||||
return self
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
"""T05-a 历史过程时间线：从历史材料提炼决策事件。
|
||||
|
||||
依据：docs/v2/14-experience-loop.md §4（加工步骤与"过程不完整"）、§3（来源定位）；
|
||||
BR-02 / E-02 / AC-02。
|
||||
|
||||
两条硬规则：
|
||||
1. **不编造经过**：只有最终报告、没有结构化问询/决定记录时，产出 **0 个事件** 并标
|
||||
`partial` + 明确提示，绝不生成虚构的问询顺序（14 §4 明文）。
|
||||
2. **不编造时间**：`occurred_at` 只来自材料里**记载**的日期；缺失或无法解析时保持
|
||||
None 并给缺证提示。**导入时间绝不冒充发生时间**（14 §3 / BR-02）。
|
||||
|
||||
提取是确定性的（别名匹配 + 结构化行），不做模型调用。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from contracts.prescreen import CaseTimeline
|
||||
from services import assistant_store as store
|
||||
|
||||
KIND = "case_timeline"
|
||||
|
||||
# 列名别名。未识别的列一律忽略——**不猜**列含义。
|
||||
COLUMN_ALIASES = {
|
||||
"occurred_at": {"date", "日期", "发生时间", "时间", "问询日期", "会议日期", "决定日期"},
|
||||
"event_type": {"event_type", "类型", "事件类型", "环节", "过程类型"},
|
||||
"question": {"question", "问题", "问询", "问询事项", "事项", "核查问题"},
|
||||
"judgment_before": {"judgment_before", "判断前", "原判断", "初审意见"},
|
||||
"judgment_after": {"judgment_after", "判断后", "新判断", "复审意见"},
|
||||
"conditions": {"conditions", "条件", "附加条件", "批注条件"},
|
||||
"evidence": {"evidence", "依据", "证据", "材料", "来源"},
|
||||
}
|
||||
|
||||
EVENT_TYPE_ALIASES = {
|
||||
"question": "question", "问询": "question", "问题": "question",
|
||||
"evidence": "evidence", "证据": "evidence", "材料": "evidence",
|
||||
"action": "action", "动作": "action", "补件": "action",
|
||||
"opinion": "opinion", "意见": "opinion", "判断": "opinion",
|
||||
"decision": "decision", "决定": "decision", "结论": "decision", "会议": "decision",
|
||||
}
|
||||
|
||||
DATE_FORMATS = ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日",
|
||||
"%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y%m%d")
|
||||
|
||||
MISSING_DATE_NOTE = "未标注发生时间，按缺证处理（不推断日期）"
|
||||
INCOMPLETE_NOTE = ("过程材料不完整：未发现结构化的问询/决定记录，"
|
||||
"仅能提炼有来源的结论，不生成虚构经过")
|
||||
|
||||
|
||||
def _normalize(text) -> str:
|
||||
return str(text or "").strip().lower().replace(" ", "")
|
||||
|
||||
|
||||
def _parse_date(value) -> datetime | None:
|
||||
"""只接受材料里明确写出的日期；无法解析即 None（不猜测）。"""
|
||||
raw = str(value or "").strip()
|
||||
if not raw or raw in ("-", "无", "null", "未知", "缺失"):
|
||||
return None
|
||||
for fmt in DATE_FORMATS:
|
||||
try:
|
||||
parsed = datetime.strptime(raw, fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
return parsed.replace(tzinfo=timezone.utc)
|
||||
try:
|
||||
parsed = datetime.fromisoformat(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _header_map(chunks: list[dict]):
|
||||
"""找到表头行及其列映射。未识别的列不猜。"""
|
||||
for position, chunk in enumerate(chunks):
|
||||
cells = chunk.get("cells")
|
||||
if not cells:
|
||||
continue
|
||||
mapping: dict[str, int] = {}
|
||||
for index, cell in enumerate(cells):
|
||||
key = _normalize(cell)
|
||||
for field, aliases in COLUMN_ALIASES.items():
|
||||
if key in aliases and field not in mapping:
|
||||
mapping[field] = index
|
||||
# 至少要能定位"问什么"或"何时"，否则不视为表头
|
||||
if "question" in mapping or "occurred_at" in mapping:
|
||||
return position, mapping
|
||||
return None, {}
|
||||
|
||||
|
||||
def _cell(cells: list, mapping: dict, field: str) -> str:
|
||||
index = mapping.get(field)
|
||||
if index is None or index >= len(cells):
|
||||
return ""
|
||||
return str(cells[index] or "").strip()
|
||||
|
||||
|
||||
def _split_conditions(text: str) -> list[str]:
|
||||
return [item.strip() for item in re.split(r"[;；,，、]", text or "") if item.strip()]
|
||||
|
||||
|
||||
def events_from_document(document: dict) -> tuple[list[dict], list[str]]:
|
||||
"""从单个文档提炼事件。返回 (events, warnings)。**不落库**。"""
|
||||
chunks = document.get("chunks") or []
|
||||
head, mapping = _header_map(chunks)
|
||||
if head is None:
|
||||
# 非结构化材料（txt/pdf/docx 正文或纯报告）→ 不产出事件，交调用方标注不完整
|
||||
return [], []
|
||||
|
||||
events, warnings = [], []
|
||||
for chunk in chunks[head + 1:]:
|
||||
cells = chunk.get("cells")
|
||||
if not cells or not any(str(cell or "").strip() for cell in cells):
|
||||
continue
|
||||
|
||||
occurred = _parse_date(_cell(cells, mapping, "occurred_at"))
|
||||
if occurred is None and "occurred_at" in mapping:
|
||||
raw = _cell(cells, mapping, "occurred_at")
|
||||
warnings.append(f"{chunk['locator']}：{MISSING_DATE_NOTE}"
|
||||
+ (f"（原文：{raw}）" if raw else "（该列留空）"))
|
||||
|
||||
raw_type = _cell(cells, mapping, "event_type")
|
||||
event_type = EVENT_TYPE_ALIASES.get(_normalize(raw_type))
|
||||
if event_type is None:
|
||||
# 未声明类型时按最保守的"问询"记；不因为缺类型而丢弃这一行
|
||||
event_type = "question"
|
||||
|
||||
question = _cell(cells, mapping, "question")
|
||||
before = _cell(cells, mapping, "judgment_before") or None
|
||||
after = _cell(cells, mapping, "judgment_after") or None
|
||||
if not (question or before or after):
|
||||
continue
|
||||
|
||||
events.append({
|
||||
"id": store.uid("event"),
|
||||
"event_type": event_type,
|
||||
"occurred_at": occurred.isoformat() if occurred else None,
|
||||
# 录入时间必填，但**只表示我们何时录入**，不当作发生时间
|
||||
"recorded_at": store.now(),
|
||||
"question": question,
|
||||
"evidence_ids": [chunk["evidence_id"]] if chunk.get("evidence_id") else [],
|
||||
"judgment_before": before,
|
||||
"judgment_after": after,
|
||||
"conditions": _split_conditions(_cell(cells, mapping, "conditions")),
|
||||
})
|
||||
return events, warnings
|
||||
|
||||
|
||||
def build(db, project_id: str, data_domain: str) -> dict:
|
||||
"""重新提炼该项目的时间线。同项目重复提炼 → 同 id、revision 递增（03 §4）。"""
|
||||
documents = store.listing(db, "document", project_id)
|
||||
events: list[dict] = []
|
||||
warnings: list[str] = []
|
||||
structured_documents = 0
|
||||
|
||||
for document in documents:
|
||||
found, notes = events_from_document(document)
|
||||
warnings.extend(notes)
|
||||
if found:
|
||||
structured_documents += 1
|
||||
events.extend(found)
|
||||
|
||||
events.sort(key=lambda item: (item["occurred_at"] is None, item["occurred_at"] or "", item["id"]))
|
||||
|
||||
if not events:
|
||||
warnings.append(INCOMPLETE_NOTE)
|
||||
status = "partial"
|
||||
elif any(event["occurred_at"] is None for event in events):
|
||||
# 设计 §4：部分缺日期为 partial
|
||||
status = "partial"
|
||||
else:
|
||||
status = "ready"
|
||||
|
||||
body = {
|
||||
# 每项目一条时间线；稳定 id 使"重新提炼"表现为 revision 递增而不是新对象
|
||||
"id": f"timeline-{project_id}",
|
||||
"project_id": project_id,
|
||||
"data_domain": data_domain,
|
||||
"status": status,
|
||||
"created_at": store.now(),
|
||||
"events": events,
|
||||
"warnings": warnings,
|
||||
}
|
||||
# 先按契约校验再落库：revision 由 store.put() 维护，这里给的 1 只是为了让
|
||||
# 校验通过（CaseTimeline.revision 必填），put() 会写入真实值。
|
||||
# 一次 put，不重复写；不合法的时间线绝不入库。
|
||||
validated = CaseTimeline(**dict(body, revision=1)).model_dump(mode="json")
|
||||
stored = store.put(db, KIND, validated)
|
||||
store.event(db, project_id, "case_timeline.built", "system",
|
||||
{"timeline_id": stored["id"], "events": len(events),
|
||||
"structured_documents": structured_documents, "status": status})
|
||||
return stored
|
||||
|
||||
|
||||
def load(db, project_id: str) -> dict:
|
||||
body = store.get(db, KIND, f"timeline-{project_id}")
|
||||
if body["project_id"] != project_id:
|
||||
raise KeyError("时间线不属于该项目")
|
||||
return body
|
||||
|
||||
|
||||
def undated_events(timeline: dict) -> list[dict]:
|
||||
"""缺发生时间的事件——界面必须显式提示缺证，不得静默排序成"近期"（BR-02）。"""
|
||||
return [event for event in timeline["events"] if event["occurred_at"] is None]
|
||||
@@ -0,0 +1,193 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user