# -*- 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()