qiaoqichaoandClaude Code
c59fff68d3
T02 第二个子卡。复用 assistant_objects（新 kind），**未新建表**；旧库结构不变。 新增服务： - services/material_import.py 批次生命周期：create/load/listing/mark_item/ retry/cancel；状态由逐项状态派生（ready/partial/failed/parsing） - services/fact_conflict.py 冲突检测/阻断/裁定；FactConflictUnresolved(E-17) 解决的**真实缺口**（BR-23 / AC-24）：既有 apply_candidate 对 values/facts 是 后写覆盖，sources 虽累积多来源但无人检测分歧、无人阻断。本层在**新路径**前置 检测：冲突时证据仍入 sources（BR-22 留痕）但值不采用（BR-23），已确认值不被覆盖。 增量改动（均为可加性，未改既有语义与调用方行为）： - assistant_store.py：新增 idempotent_lookup / idempotent_remember 共享 helper。 既有 enqueue 与 create_report 的内联实现**未改写**。用 INSERT OR REPLACE 而非 裸 INSERT：重试路径上裸 INSERT 会抛 sqlite3.IntegrityError（非 Conflict， translate() 无法映射为 409）。 - assistant_materials.apply_candidate 新增 adopt=True 形参。adopt=False 只记证据 不改值；既有调用方不传参，行为完全不变（有回归测试）。 设计决定： - T00/04 曾提 3 个 kind；实现时发现契约中 ImportItem 无 revision 且内嵌于批次， items 是值对象，故收敛为 2 个 kind（import_batch / fact_conflict） - E-17 有两个面：确认动作被拒 = 422；运行级结果 = HTTP 200 且冲突存于 degradations。契约映射与注释均已写明，并有测试锁定 测试：新增 test_prescreen_storage.py（33 项）。全量 196 项 OK（124 → 196）。 scripts/verify.py 9/9 PASS（exit 0）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
442 lines
23 KiB
Python
442 lines
23 KiB
Python
"""S1 存储层测试：材料导入批次与事实冲突。
|
||
|
||
重点证明的是**旧实现会失败的行为**：`apply_candidate` 对字段值是后写覆盖，
|
||
本层必须让"后到的冲突来源不得覆盖已确认值"成立（BR-23 / AC-24）。
|
||
|
||
用独立临时 SQLite 库，绝不触碰 backend/xingbang_data 下的真实库。
|
||
"""
|
||
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from contracts.prescreen import ErrorCode
|
||
from services import assistant_store as store
|
||
from services import fact_conflict, material_import
|
||
from services.assistant_materials import FIELDS
|
||
|
||
|
||
def manifest_entry(name="a.csv", sha="a" * 64, size=100, mime="text/csv"):
|
||
return {"id": store.uid("item"), "relative_path": name, "sha256": sha,
|
||
"size_bytes": size, "mime_type": mime, "status": "queued",
|
||
"received_at": store.now()}
|
||
|
||
|
||
class StoreTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self._tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = Path(self._tmp.name) / "prescreen_test.sqlite3"
|
||
store.initialize(self.db_path)
|
||
self.actor = "demo-manager"
|
||
with store.transaction(self.db_path) as db:
|
||
self._seed_project(db)
|
||
|
||
def tearDown(self):
|
||
self._tmp.cleanup()
|
||
|
||
@property
|
||
def project_id(self):
|
||
return "DEMO-A"
|
||
|
||
def _seed_project(self, db):
|
||
store.put(db, "project", {
|
||
"id": self.project_id, "data_domain": "demo", "revision": 1,
|
||
"facts": {}, "values": {}, "sources": {}, "selected_sources": {},
|
||
"document_ids": [], "pending_candidates": [], "data_revision": 1,
|
||
})
|
||
|
||
def project(self, db):
|
||
return store.get(db, "project", self.project_id)
|
||
|
||
def candidate(self, field="facility.amount", value="1000", unit="万元", **over):
|
||
row = {"id": store.uid("candidate"), "field": field, "value": value,
|
||
"unit": unit, "currency": "CNY", "period": "2025", "scope": "合并",
|
||
"source": "a.csv", "locator": "第2行", "document_id": "doc-1"}
|
||
row.update(over)
|
||
return row
|
||
|
||
|
||
class ImportBatchTests(StoreTestCase):
|
||
def test_create_is_queued_with_derived_total(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(
|
||
db, self.project_id, "demo",
|
||
[manifest_entry(size=100), manifest_entry(name="b.csv", sha="b" * 64, size=50)],
|
||
self.actor)
|
||
self.assertEqual(batch["status"], "queued")
|
||
self.assertEqual(batch["total_bytes"], 150)
|
||
self.assertEqual(batch["revision"], 1)
|
||
self.assertEqual(batch["parser_version"], material_import.PARSER_VERSION)
|
||
|
||
def test_empty_manifest_rejected(self):
|
||
with store.transaction(self.db_path) as db:
|
||
with self.assertRaises(ValueError) as ctx:
|
||
material_import.create(db, self.project_id, "demo", [], self.actor)
|
||
self.assertIn("为空", str(ctx.exception))
|
||
|
||
def test_duplicate_sha_in_one_batch_rejected(self):
|
||
with store.transaction(self.db_path) as db:
|
||
with self.assertRaises(ValueError) as ctx:
|
||
material_import.create(db, self.project_id, "demo",
|
||
[manifest_entry(), manifest_entry(name="b.csv")], self.actor)
|
||
self.assertIn("哈希重复", str(ctx.exception))
|
||
|
||
def test_absolute_path_entry_reports_which_item(self):
|
||
with store.transaction(self.db_path) as db:
|
||
with self.assertRaises(ValueError) as ctx:
|
||
material_import.create(db, self.project_id, "demo",
|
||
[manifest_entry(name="C:/x.csv")], self.actor)
|
||
self.assertIn("第 1 个文件", str(ctx.exception))
|
||
|
||
def test_batch_total_over_limit_rejected(self):
|
||
from contracts.prescreen import MAX_ITEM_BYTES
|
||
entries = [manifest_entry(name=f"f{n}.csv", sha=f"{n}" * 64, size=MAX_ITEM_BYTES)
|
||
for n in range(3)]
|
||
with store.transaction(self.db_path) as db:
|
||
with self.assertRaises(ValueError) as ctx:
|
||
material_import.create(db, self.project_id, "demo", entries, self.actor)
|
||
self.assertIn("超过上限", str(ctx.exception))
|
||
|
||
def test_idempotent_replay_returns_same_batch(self):
|
||
entries = [manifest_entry()]
|
||
with store.transaction(self.db_path) as db:
|
||
first = material_import.create(db, self.project_id, "demo", entries,
|
||
self.actor, idempotency_key="k1")
|
||
again = material_import.create(db, self.project_id, "demo", entries,
|
||
self.actor, idempotency_key="k1")
|
||
self.assertEqual(first["id"], again["id"])
|
||
|
||
def test_idempotent_key_with_different_payload_conflicts(self):
|
||
with store.transaction(self.db_path) as db:
|
||
material_import.create(db, self.project_id, "demo", [manifest_entry()],
|
||
self.actor, idempotency_key="k2")
|
||
with self.assertRaises(store.Conflict):
|
||
material_import.create(db, self.project_id, "demo",
|
||
[manifest_entry(sha="c" * 64)], self.actor,
|
||
idempotency_key="k2")
|
||
|
||
def test_status_derivation_partial_ready_failed(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(
|
||
db, self.project_id, "demo",
|
||
[manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)],
|
||
self.actor)
|
||
ids = [item["id"] for item in batch["items"]]
|
||
batch = material_import.mark_item(db, batch["id"], ids[0], "parsed",
|
||
document_id="doc-1")
|
||
self.assertEqual(batch["status"], "parsing")
|
||
batch = material_import.mark_item(db, batch["id"], ids[1], "failed",
|
||
error_code=ErrorCode.MATERIAL_PARSE_FAILED)
|
||
self.assertEqual(batch["status"], "partial")
|
||
self.assertEqual(batch["revision"], 3)
|
||
|
||
def test_mark_item_failed_requires_error_code(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor)
|
||
with self.assertRaises(ValueError):
|
||
material_import.mark_item(db, batch["id"], batch["items"][0]["id"], "failed")
|
||
|
||
def test_retry_only_resets_failed_items(self):
|
||
# AC-23 反例：已 parsed 的条目不得被重解析、不得被重写
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(
|
||
db, self.project_id, "demo",
|
||
[manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)],
|
||
self.actor)
|
||
ids = [item["id"] for item in batch["items"]]
|
||
material_import.mark_item(db, batch["id"], ids[0], "parsed", document_id="doc-1")
|
||
batch = material_import.mark_item(db, batch["id"], ids[1], "failed",
|
||
error_code=ErrorCode.MATERIAL_PARSE_FAILED)
|
||
parsed_before = next(i for i in batch["items"] if i["id"] == ids[0])
|
||
|
||
batch = material_import.retry(db, batch["id"], [ids[1]], self.actor,
|
||
expected=batch["revision"])
|
||
parsed_after = next(i for i in batch["items"] if i["id"] == ids[0])
|
||
self.assertEqual(parsed_before, parsed_after)
|
||
self.assertEqual(next(i for i in batch["items"] if i["id"] == ids[1])["status"], "queued")
|
||
|
||
def test_retry_rejects_non_failed_item(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(
|
||
db, self.project_id, "demo",
|
||
[manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)],
|
||
self.actor)
|
||
ids = [item["id"] for item in batch["items"]]
|
||
material_import.mark_item(db, batch["id"], ids[0], "parsed", document_id="doc-1")
|
||
batch = material_import.mark_item(db, batch["id"], ids[1], "failed",
|
||
error_code=ErrorCode.MATERIAL_PARSE_FAILED)
|
||
with self.assertRaises(ValueError):
|
||
material_import.retry(db, batch["id"], [ids[0]], self.actor, expected=batch["revision"])
|
||
|
||
def test_retry_with_stale_revision_conflicts(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(db, self.project_id, "demo",
|
||
[manifest_entry()], self.actor)
|
||
material_import.mark_item(db, batch["id"], batch["items"][0]["id"], "failed",
|
||
error_code=ErrorCode.MATERIAL_PARSE_FAILED)
|
||
with self.assertRaises(store.Conflict):
|
||
material_import.retry(db, batch["id"], [batch["items"][0]["id"]],
|
||
self.actor, expected=1)
|
||
|
||
def test_cancel_then_late_result_is_ignored(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor)
|
||
batch = material_import.cancel(db, batch["id"], self.actor, expected=batch["revision"])
|
||
item_id = batch["items"][0]["id"]
|
||
after = material_import.mark_item(db, batch["id"], item_id, "parsed", document_id="doc-1")
|
||
self.assertEqual(after["status"], "cancelled")
|
||
self.assertIsNone(after["items"][0]["document_id"])
|
||
|
||
def test_cross_project_read_is_not_found(self):
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor)
|
||
with self.assertRaises(KeyError):
|
||
material_import.load(db, batch["id"], project_id="DEMO-B")
|
||
|
||
|
||
class FactConflictTests(StoreTestCase):
|
||
def _confirm(self, db, candidate):
|
||
"""走一遍真实链路：detect 就地改 project → 保存 → 有冲突则落库。
|
||
|
||
注意必须先取到 project 字典再保存**同一个对象**：detect 是就地修改，
|
||
若重新 store.get 一份新副本去 put，改动会被丢弃。
|
||
"""
|
||
project = self.project(db)
|
||
payload = fact_conflict.detect(project, candidate, self.actor, "初始确认")
|
||
store.put(db, "project", project)
|
||
if payload:
|
||
return fact_conflict.create(db, self.project_id, payload, self.actor)
|
||
return None
|
||
|
||
def _resolve(self, db, conflict, selected, note="裁定说明"):
|
||
project = self.project(db)
|
||
row = fact_conflict.resolve(db, conflict["id"], project, selected, note,
|
||
self.actor, expected=conflict["revision"])
|
||
store.put(db, "project", project)
|
||
return row
|
||
|
||
def _confirmed_value(self, project, field):
|
||
return project["values"].get(field)
|
||
|
||
def test_first_confirmation_has_no_conflict(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self.assertIsNone(self._confirm(db, self.candidate(value="1000")))
|
||
project = self.project(db)
|
||
# 不硬编码换算结果：断言"值等于被采纳证据的值"，与单元/币种换算解耦
|
||
evidence = project["sources"]["facility.amount"][-1]
|
||
self.assertEqual(self._confirmed_value(project, "facility.amount"), evidence["value"])
|
||
self.assertIsNotNone(self._confirmed_value(project, "facility.amount"))
|
||
|
||
def test_conflicting_second_source_does_not_overwrite_and_records_evidence(self):
|
||
# ★ 这是旧实现会失败的核心行为：apply_candidate 曾直接覆盖
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
before = self._confirmed_value(self.project(db), "facility.amount")
|
||
|
||
conflict = self._confirm(db, self.candidate(value="1200", source="b.csv", locator="第9行"))
|
||
self.assertIsNotNone(conflict)
|
||
self.assertEqual(conflict["status"], "unresolved")
|
||
self.assertEqual(conflict["field"], "facility.amount")
|
||
|
||
project = self.project(db)
|
||
self.assertEqual(self._confirmed_value(project, "facility.amount"), before,
|
||
"已确认值被覆盖了")
|
||
# 证据仍按 BR-22 留痕：两个来源都在
|
||
self.assertEqual(len(project["sources"]["facility.amount"]), 2)
|
||
self.assertEqual(len(conflict["candidates"]), 2)
|
||
self.assertNotEqual(conflict["candidates"][0]["value"],
|
||
conflict["candidates"][1]["value"])
|
||
|
||
def test_blocked_field_refuses_confirmation(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
self._confirm(db, self.candidate(value="1200"))
|
||
with self.assertRaises(fact_conflict.FactConflictUnresolved) as ctx:
|
||
fact_conflict.assert_confirmable(db, self.project_id, ["facility.amount"])
|
||
self.assertIn("不一致", str(ctx.exception))
|
||
self.assertEqual(ctx.exception.code, ErrorCode.FACT_CONFLICT_UNRESOLVED)
|
||
|
||
def test_other_fields_remain_confirmable(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
self._confirm(db, self.candidate(value="1200"))
|
||
fact_conflict.assert_confirmable(db, self.project_id, ["financial.total_assets"])
|
||
|
||
def test_same_value_is_not_a_conflict(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
self.assertIsNone(self._confirm(db, self.candidate(value="1000", source="b.csv")))
|
||
self.assertEqual(fact_conflict.blocked_fields(db, self.project_id), set())
|
||
|
||
def test_unknown_candidate_does_not_create_conflict(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
self.assertIsNone(self._confirm(db, self.candidate(value=None)))
|
||
|
||
def test_resolve_adopts_chosen_value_and_unblocks(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
chosen = conflict["candidates"][1]["evidence_id"]
|
||
|
||
resolved = self._resolve(db, conflict, [chosen], "以合同金额为准")
|
||
self.assertEqual(resolved["status"], "resolved")
|
||
self.assertEqual(resolved["resolved_by"], self.actor)
|
||
self.assertIsNotNone(resolved["resolved_at"])
|
||
|
||
project = self.project(db)
|
||
# 采纳后该字段的值来自被选中的候选
|
||
self.assertEqual(str(self._confirmed_value(project, "facility.amount")),
|
||
str(conflict["candidates"][1]["value"]))
|
||
self.assertEqual(project["selected_sources"]["facility.amount"], chosen)
|
||
self.assertEqual(fact_conflict.blocked_fields(db, self.project_id), set())
|
||
|
||
def test_resolve_can_keep_the_original_value(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
original = conflict["candidates"][0]["evidence_id"]
|
||
self._resolve(db, conflict, [original], "以已确认为准")
|
||
project = self.project(db)
|
||
self.assertEqual(str(self._confirmed_value(project, "facility.amount")),
|
||
str(conflict["candidates"][0]["value"]))
|
||
|
||
def test_resolve_requires_note_and_valid_selection(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
chosen = conflict["candidates"][0]["evidence_id"]
|
||
|
||
with self.assertRaises(ValueError):
|
||
fact_conflict.resolve(db, conflict["id"], self.project(db), [chosen], " ",
|
||
self.actor, expected=conflict["revision"])
|
||
with self.assertRaises(ValueError) as ctx:
|
||
fact_conflict.resolve(db, conflict["id"], self.project(db), ["ev-999"],
|
||
"说明", self.actor, expected=conflict["revision"])
|
||
self.assertIn("不属于", str(ctx.exception))
|
||
with self.assertRaises(ValueError):
|
||
fact_conflict.resolve(db, conflict["id"], self.project(db), [],
|
||
"说明", self.actor, expected=conflict["revision"])
|
||
# 三次被拒后冲突仍未被改动
|
||
self.assertEqual(fact_conflict.load(db, conflict["id"])["status"], "unresolved")
|
||
|
||
def test_double_resolve_conflicts(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
chosen = conflict["candidates"][1]["evidence_id"]
|
||
self._resolve(db, conflict, [chosen])
|
||
with self.assertRaises(store.Conflict):
|
||
fact_conflict.resolve(db, conflict["id"], self.project(db), [chosen], "再裁一次",
|
||
self.actor, expected=conflict["revision"])
|
||
|
||
def test_third_source_creates_new_conflict_with_supersedes(self):
|
||
# AC-24：再次冲突新建对象并指向旧对象，不覆写历史
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
first = self._confirm(db, self.candidate(value="1200"))
|
||
self._resolve(db, first, [first["candidates"][1]["evidence_id"]], "采纳1200")
|
||
|
||
project = self.project(db)
|
||
payload = fact_conflict.detect(project, self.candidate(value="900", source="c.csv"),
|
||
self.actor, "第三来源")
|
||
store.put(db, "project", project)
|
||
self.assertIsNotNone(payload, "第三来源与已确认值不同，应产生冲突")
|
||
second = fact_conflict.create(db, self.project_id, payload, self.actor,
|
||
supersedes_id=first["id"])
|
||
self.assertEqual(second["supersedes_id"], first["id"])
|
||
self.assertNotEqual(second["id"], first["id"])
|
||
# 旧记录仍为 resolved，未被改写
|
||
old = fact_conflict.load(db, first["id"])
|
||
self.assertEqual(old["status"], "resolved")
|
||
self.assertEqual(old["resolution_note"], "采纳1200")
|
||
|
||
def test_non_standard_field_rejected(self):
|
||
with store.transaction(self.db_path) as db:
|
||
with self.assertRaises(ValueError) as ctx:
|
||
fact_conflict.detect(self.project(db),
|
||
self.candidate(field="not.a.real.field"), self.actor, "x")
|
||
self.assertIn("不支持的字段", str(ctx.exception))
|
||
|
||
def test_legacy_path_still_adopts_value(self):
|
||
# apply_candidate 的既有行为不受影响：仍然正常采用值
|
||
from services.assistant_materials import apply_candidate
|
||
with store.transaction(self.db_path) as db:
|
||
project = self.project(db)
|
||
evidence = apply_candidate(project, self.candidate(value="1000"), self.actor, "旧路径")
|
||
self.assertEqual(project["values"]["facility.amount"], evidence["value"])
|
||
self.assertEqual(project["selected_sources"]["facility.amount"], evidence["id"])
|
||
|
||
def test_adopt_false_records_evidence_without_adopting(self):
|
||
from services.assistant_materials import apply_candidate
|
||
with store.transaction(self.db_path) as db:
|
||
project = self.project(db)
|
||
evidence = apply_candidate(project, self.candidate(value="1000"), self.actor,
|
||
"待裁定", adopt=False)
|
||
self.assertNotIn("facility.amount", project["values"])
|
||
self.assertNotIn("facility.amount", project["facts"])
|
||
self.assertEqual(len(project["sources"]["facility.amount"]), 1)
|
||
self.assertEqual(project["sources"]["facility.amount"][0]["id"], evidence["id"])
|
||
|
||
def test_listing_filters_by_status(self):
|
||
with store.transaction(self.db_path) as db:
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
self.assertEqual(
|
||
len(fact_conflict.listing(db, self.project_id, status="unresolved")["items"]), 1)
|
||
self.assertEqual(
|
||
len(fact_conflict.listing(db, self.project_id, status="resolved")["items"]), 0)
|
||
self._resolve(db, conflict, [conflict["candidates"][1]["evidence_id"]])
|
||
self.assertEqual(
|
||
len(fact_conflict.listing(db, self.project_id, status="resolved")["items"]), 1)
|
||
self.assertEqual(
|
||
len(fact_conflict.listing(db, self.project_id, status="unresolved")["items"]), 0)
|
||
|
||
def test_standard_field_whitelist_is_the_existing_one(self):
|
||
# 不复制白名单：以 assistant_materials.FIELDS 为唯一来源
|
||
self.assertIn("facility.amount", fact_conflict.STANDARD_FIELDS)
|
||
self.assertEqual(fact_conflict.STANDARD_FIELDS, frozenset(FIELDS.values()))
|
||
|
||
def test_objects_are_persisted_through_the_generic_store(self):
|
||
"""复用 assistant_objects（新 kind），未新建表 —— DDL 结论的机械证明。"""
|
||
with store.transaction(self.db_path) as db:
|
||
batch = material_import.create(db, self.project_id, "demo",
|
||
[manifest_entry()], self.actor)
|
||
self._confirm(db, self.candidate(value="1000"))
|
||
conflict = self._confirm(db, self.candidate(value="1200"))
|
||
|
||
kinds = {row[0] for row in db.execute(
|
||
"SELECT DISTINCT kind FROM assistant_objects").fetchall()}
|
||
self.assertIn("import_batch", kinds)
|
||
self.assertIn("fact_conflict", kinds)
|
||
self.assertEqual(conflict["revision"], 1)
|
||
self.assertEqual(batch["revision"], 1)
|
||
|
||
# 表结构未变：仍是 S1 之前的那 5 张表
|
||
tables = {row[0] for row in db.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
|
||
self.assertTrue({"assistant_objects", "assistant_events", "assistant_idempotency",
|
||
"assistant_jobs", "assistant_meta"} <= tables)
|
||
|
||
|
||
class IsolationTests(StoreTestCase):
|
||
def test_services_do_not_touch_the_real_data_root(self):
|
||
# 本模块的测试库必须在临时目录；真实库不受影响
|
||
self.assertNotIn("xingbang_data", str(self.db_path))
|
||
self.assertTrue(str(self.db_path).startswith(tempfile.gettempdir().replace("\\", "/"))
|
||
or "Temp" in str(self.db_path) or "tmp" in str(self.db_path))
|
||
|
||
def test_contracts_module_not_imported_by_services_is_false_direction_guard(self):
|
||
# 方向守卫：contracts 不得导入 services（本测试从 services 侧反查）
|
||
import contracts.prescreen as module
|
||
with open(module.__file__, encoding="utf-8") as handle:
|
||
source = handle.read()
|
||
self.assertNotIn("from services", source)
|
||
self.assertNotIn("import services", source)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|