275 lines
10 KiB
Plaintext
275 lines
10 KiB
Plaintext
# -*- coding: utf-8 -*-
|
||
# 样例：复制到 ``tests/test_sync_contract.py`` 后按实际 domain service 改名运行。
|
||
# 默认不会被 ``python tests/run_tests.py`` 收集（扩展名为 .sample）。
|
||
"""
|
||
外源 → 本地库同步契约范式（通用，无具体平台业务名）。
|
||
|
||
覆盖：
|
||
- 首次同步插入
|
||
- 第二次同步幂等
|
||
- 更新已有数据
|
||
- 处理失效数据
|
||
- 不可信空结果保护 vs 完整空快照失效（``snapshot_complete`` 必填，无默认值）
|
||
- 遗漏 ``snapshot_complete`` 立即失败（TypeError）
|
||
- 事务在部分写入后失败并回滚
|
||
- 返回 inserted/updated/unchanged/removed
|
||
- sync Action 不调用 CSV/Excel 文件导出器
|
||
|
||
复制后把内存仓库与 stub 外源换成真实 repository / reader；禁止访问真实网络或用户数据目录。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
|
||
@dataclass
|
||
class SyncStats:
|
||
total: int = 0
|
||
inserted: int = 0
|
||
updated: int = 0
|
||
unchanged: int = 0
|
||
removed: int = 0
|
||
|
||
def as_dict(self) -> dict[str, int]:
|
||
return {
|
||
"total": self.total,
|
||
"inserted": self.inserted,
|
||
"updated": self.updated,
|
||
"unchanged": self.unchanged,
|
||
"removed": self.removed,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class MemoryRecordStore:
|
||
"""内存库：演示幂等 upsert + 全量失效（is_active 标记）。"""
|
||
|
||
rows: dict[str, dict[str, Any]] = field(default_factory=dict)
|
||
in_transaction: bool = False
|
||
_tx_snapshot: dict[str, dict[str, Any]] | None = None
|
||
|
||
def begin(self) -> None:
|
||
self._tx_snapshot = {k: dict(v) for k, v in self.rows.items()}
|
||
self.in_transaction = True
|
||
|
||
def commit(self) -> None:
|
||
self._tx_snapshot = None
|
||
self.in_transaction = False
|
||
|
||
def rollback(self) -> None:
|
||
if self._tx_snapshot is not None:
|
||
self.rows = {k: dict(v) for k, v in self._tx_snapshot.items()}
|
||
self._tx_snapshot = None
|
||
self.in_transaction = False
|
||
|
||
def upsert_snapshot(
|
||
self,
|
||
records: list[dict[str, Any]],
|
||
*,
|
||
fail_after_writes: int | None = None,
|
||
) -> SyncStats:
|
||
"""全量快照：按 business_key 对齐；缺失标 is_active=0。
|
||
|
||
``fail_after_writes``：在成功完成该次数的 insert/update 之后抛错，用于验证回滚。
|
||
"""
|
||
self.begin()
|
||
try:
|
||
stats = SyncStats(total=len(records))
|
||
seen: set[str] = set()
|
||
write_count = 0
|
||
for rec in records:
|
||
key = str(rec["business_key"])
|
||
seen.add(key)
|
||
existing = self.rows.get(key)
|
||
payload = {
|
||
"business_key": key,
|
||
"title": rec["title"],
|
||
"is_active": 1,
|
||
}
|
||
if existing is None:
|
||
self.rows[key] = payload
|
||
stats.inserted += 1
|
||
write_count += 1
|
||
elif existing.get("title") == payload["title"] and existing.get("is_active") == 1:
|
||
stats.unchanged += 1
|
||
else:
|
||
self.rows[key] = payload
|
||
stats.updated += 1
|
||
write_count += 1
|
||
if fail_after_writes is not None and write_count >= fail_after_writes:
|
||
raise RuntimeError(
|
||
f"simulated write failure after {write_count} insert/update write(s)"
|
||
)
|
||
for key, existing in list(self.rows.items()):
|
||
if key not in seen and existing.get("is_active") == 1:
|
||
existing["is_active"] = 0
|
||
stats.removed += 1
|
||
self.commit()
|
||
return stats
|
||
except Exception:
|
||
self.rollback()
|
||
raise
|
||
|
||
|
||
class ExportSpy:
|
||
"""若 sync 路径误调导出器，测试应失败。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.calls: list[str] = []
|
||
|
||
def write_csv(self, *_a: Any, **_k: Any) -> None:
|
||
self.calls.append("csv")
|
||
|
||
def write_excel(self, *_a: Any, **_k: Any) -> None:
|
||
self.calls.append("excel")
|
||
|
||
|
||
def sync_records(
|
||
store: MemoryRecordStore,
|
||
external_rows: list[dict[str, Any]],
|
||
*,
|
||
snapshot_complete: bool,
|
||
exporter: ExportSpy | None = None,
|
||
fail_after_writes: int | None = None,
|
||
) -> dict[str, int]:
|
||
"""领域同步内核示意：只写库，不导出文件。
|
||
|
||
``snapshot_complete`` 为 keyword-only **必填**参数（无默认值）：
|
||
- False：读取失败/分页不完整/不可信结果 → 不得清库或失效旧数据
|
||
- True：外源完整快照（可为空）→ 允许按全量策略 upsert / 标记失效
|
||
调用方必须显式传入，禁止默认当作完整快照，也禁止靠条数猜测完整性。
|
||
"""
|
||
if external_rows is None:
|
||
raise ValueError("external_rows required")
|
||
if not snapshot_complete:
|
||
active = sum(1 for r in store.rows.values() if r.get("is_active") == 1)
|
||
return SyncStats(total=0, unchanged=active).as_dict()
|
||
stats = store.upsert_snapshot(external_rows, fail_after_writes=fail_after_writes)
|
||
if exporter is not None:
|
||
# 正确实现不得调用；本示意刻意不调用，供测试断言
|
||
pass
|
||
return stats.as_dict()
|
||
|
||
|
||
class TestSyncContractSample(unittest.TestCase):
|
||
def test_first_sync_inserts(self) -> None:
|
||
store = MemoryRecordStore()
|
||
stats = sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A"}, {"business_key": "b", "title": "B"}],
|
||
snapshot_complete=True,
|
||
)
|
||
self.assertEqual(stats["inserted"], 2)
|
||
self.assertEqual(stats["total"], 2)
|
||
self.assertEqual(len(store.rows), 2)
|
||
|
||
def test_second_sync_is_idempotent(self) -> None:
|
||
store = MemoryRecordStore()
|
||
rows = [{"business_key": "a", "title": "A"}]
|
||
sync_records(store, rows, snapshot_complete=True)
|
||
stats = sync_records(store, rows, snapshot_complete=True)
|
||
self.assertEqual(stats["unchanged"], 1)
|
||
self.assertEqual(stats["inserted"], 0)
|
||
self.assertEqual(stats["updated"], 0)
|
||
|
||
def test_updates_existing_row(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(store, [{"business_key": "a", "title": "A"}], snapshot_complete=True)
|
||
stats = sync_records(
|
||
store, [{"business_key": "a", "title": "A2"}], snapshot_complete=True
|
||
)
|
||
self.assertEqual(stats["updated"], 1)
|
||
self.assertEqual(store.rows["a"]["title"], "A2")
|
||
|
||
def test_marks_removed_inactive_on_full_sync(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A"}, {"business_key": "b", "title": "B"}],
|
||
snapshot_complete=True,
|
||
)
|
||
stats = sync_records(
|
||
store, [{"business_key": "a", "title": "A"}], snapshot_complete=True
|
||
)
|
||
self.assertEqual(stats["removed"], 1)
|
||
self.assertEqual(store.rows["b"]["is_active"], 0)
|
||
self.assertEqual(store.rows["a"]["is_active"], 1)
|
||
|
||
def test_incomplete_empty_snapshot_does_not_wipe_existing(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(store, [{"business_key": "a", "title": "A"}], snapshot_complete=True)
|
||
stats = sync_records(store, [], snapshot_complete=False)
|
||
self.assertEqual(store.rows["a"]["is_active"], 1)
|
||
self.assertEqual(stats["removed"], 0)
|
||
self.assertEqual(stats["unchanged"], 1)
|
||
|
||
def test_complete_empty_snapshot_marks_existing_inactive(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A"}, {"business_key": "b", "title": "B"}],
|
||
snapshot_complete=True,
|
||
)
|
||
stats = sync_records(store, [], snapshot_complete=True)
|
||
self.assertEqual(stats["removed"], 2)
|
||
self.assertEqual(store.rows["a"]["is_active"], 0)
|
||
self.assertEqual(store.rows["b"]["is_active"], 0)
|
||
|
||
def test_omitting_snapshot_complete_raises_type_error(self) -> None:
|
||
store = MemoryRecordStore()
|
||
with self.assertRaises(TypeError):
|
||
sync_records(store, [{"business_key": "a", "title": "A"}]) # type: ignore[call-arg]
|
||
|
||
def test_transaction_failure_after_partial_writes_rolls_back(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(store, [{"business_key": "a", "title": "A"}], snapshot_complete=True)
|
||
with self.assertRaises(RuntimeError) as ctx:
|
||
sync_records(
|
||
store,
|
||
[
|
||
{"business_key": "a", "title": "A2"},
|
||
{"business_key": "b", "title": "B"},
|
||
],
|
||
snapshot_complete=True,
|
||
fail_after_writes=1,
|
||
)
|
||
self.assertIn("after 1", str(ctx.exception))
|
||
self.assertEqual(store.rows["a"]["title"], "A")
|
||
self.assertNotIn("b", store.rows)
|
||
self.assertFalse(store.in_transaction)
|
||
self.assertIsNone(store._tx_snapshot)
|
||
|
||
def test_stats_keys_cover_inserted_updated_unchanged_removed(self) -> None:
|
||
store = MemoryRecordStore()
|
||
sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A"}, {"business_key": "b", "title": "B"}],
|
||
snapshot_complete=True,
|
||
)
|
||
stats = sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A2"}],
|
||
snapshot_complete=True,
|
||
)
|
||
for key in ("total", "inserted", "updated", "unchanged", "removed"):
|
||
self.assertIn(key, stats)
|
||
self.assertEqual(stats["updated"], 1)
|
||
self.assertEqual(stats["removed"], 1)
|
||
|
||
def test_sync_does_not_call_file_exporters(self) -> None:
|
||
store = MemoryRecordStore()
|
||
spy = ExportSpy()
|
||
sync_records(
|
||
store,
|
||
[{"business_key": "a", "title": "A"}],
|
||
snapshot_complete=True,
|
||
exporter=spy,
|
||
)
|
||
self.assertEqual(spy.calls, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|