# -*- coding: utf-8 -*-
"""ODS 去重与软删除标准化 — 属性测试。

Feature: ods-dedup-standardize
使用 hypothesis 验证 OdsTaskSpec 配置校验的正确性属性。
"""
from __future__ import annotations

import os
import sys
from pathlib import Path

import pytest
from hypothesis import given, settings, HealthCheck
from hypothesis import strategies as st

PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

os.environ.setdefault("ETL_SKIP_DOTENV", "1")

from tasks.ods.ods_tasks import OdsTaskSpec, SnapshotMode

# ---------------------------------------------------------------------------
# 固定必填字段，测试只关注 snapshot_mode / snapshot_time_column 组合
# ---------------------------------------------------------------------------
_FIXED = dict(
    code="TEST_TASK",
    class_name="TestTask",
    table_name="ods.test_table",
    endpoint="/test",
)


# ---------------------------------------------------------------------------
# Property 1: SnapshotMode 与 snapshot_time_column 一致性
# **Validates: Requirements 2.3, 2.4, 2.5, 2.6**
# ---------------------------------------------------------------------------

class TestProperty1SnapshotModeConsistency:
    """Property 1: SnapshotMode 与 snapshot_time_column 一致性。

    *For any* OdsTaskSpec 实例，当 snapshot_mode 为 WINDOW 时
    snapshot_time_column 必须为非空字符串；当 snapshot_mode 为
    FULL_TABLE 或 NONE 时 snapshot_time_column 必须为 None。
    违反此约束应抛出 ValueError。
    """

    # --- WINDOW 模式 ---

    @given(col=st.text(min_size=1))
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_1_window_with_nonempty_column_succeeds(self, col: str):
        """WINDOW + 非空 snapshot_time_column → 构造成功。"""
        spec = OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.WINDOW, snapshot_time_column=col)
        assert spec.snapshot_mode == SnapshotMode.WINDOW
        assert spec.snapshot_time_column == col

    @given(data=st.just(None))
    @settings(max_examples=10)
    def test_property_1_window_with_none_column_raises(self, data):
        """WINDOW + None → ValueError。"""
        with pytest.raises(ValueError, match="snapshot_time_column"):
            OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.WINDOW, snapshot_time_column=None)

    # --- FULL_TABLE 模式 ---

    @given(data=st.just(None))
    @settings(max_examples=10)
    def test_property_1_full_table_with_none_column_succeeds(self, data):
        """FULL_TABLE + None → 构造成功。"""
        spec = OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.FULL_TABLE, snapshot_time_column=None)
        assert spec.snapshot_mode == SnapshotMode.FULL_TABLE
        assert spec.snapshot_time_column is None

    @given(col=st.text(min_size=1))
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_1_full_table_with_nonempty_column_raises(self, col: str):
        """FULL_TABLE + 非空字符串 → ValueError。"""
        with pytest.raises(ValueError, match="snapshot_time_column"):
            OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.FULL_TABLE, snapshot_time_column=col)

    # --- NONE 模式 ---

    @given(data=st.just(None))
    @settings(max_examples=10)
    def test_property_1_none_mode_with_none_column_succeeds(self, data):
        """NONE + None → 构造成功。"""
        spec = OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.NONE, snapshot_time_column=None)
        assert spec.snapshot_mode == SnapshotMode.NONE
        assert spec.snapshot_time_column is None

    @given(col=st.text(min_size=1))
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_1_none_mode_with_nonempty_column_raises(self, col: str):
        """NONE + 非空字符串 → ValueError。"""
        with pytest.raises(ValueError, match="snapshot_time_column"):
            OdsTaskSpec(**_FIXED, snapshot_mode=SnapshotMode.NONE, snapshot_time_column=col)

    # --- 综合随机组合 ---

    @given(
        mode=st.sampled_from(SnapshotMode),
        col=st.one_of(st.none(), st.text(min_size=1)),
    )
    @settings(max_examples=200, suppress_health_check=[HealthCheck.too_slow])
    def test_property_1_random_combinations(self, mode: SnapshotMode, col: str | None):
        """随机 SnapshotMode × snapshot_time_column 组合，验证校验不变量。"""
        should_succeed = (
            (mode == SnapshotMode.WINDOW and col is not None and len(col) > 0)
            or (mode != SnapshotMode.WINDOW and col is None)
        )
        if should_succeed:
            spec = OdsTaskSpec(**_FIXED, snapshot_mode=mode, snapshot_time_column=col)
            assert spec.snapshot_mode == mode
            assert spec.snapshot_time_column == col
        else:
            with pytest.raises(ValueError):
                OdsTaskSpec(**_FIXED, snapshot_mode=mode, snapshot_time_column=col)


from tasks.ods.ods_tasks import BaseOdsTask

# ---------------------------------------------------------------------------
# Property 2: content_hash 确定性
# **Validates: Requirements 5.1, 5.4**
# ---------------------------------------------------------------------------

class TestProperty2ContentHashDeterminism:
    """Property 2: content_hash 确定性。

    *For any* 原始 payload（合法 JSON 对象）和 is_delete 值（0 或 1），
    对相同的 (payload, is_delete) 输入调用 _compute_content_hash
    应始终产生相同的 SHA-256 哈希值。
    """

    @given(
        payload=st.dictionaries(st.text(), st.text()),
        is_delete=st.sampled_from([0, 1]),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_2_same_input_same_hash(self, payload: dict, is_delete: int):
        """相同 (payload, is_delete) 两次调用产生相同 hash。"""
        h1 = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=is_delete)
        h2 = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=is_delete)
        assert h1 == h2, f"确定性违反: 相同输入产生不同 hash: {h1!r} != {h2!r}"

    @given(
        payload=st.dictionaries(st.text(), st.text()),
        is_delete=st.sampled_from([0, 1]),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_2_hash_is_valid_sha256(self, payload: dict, is_delete: int):
        """返回值是合法的 64 字符十六进制 SHA-256 哈希。"""
        h = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=is_delete)
        assert isinstance(h, str)
        assert len(h) == 64, f"SHA-256 应为 64 字符，实际 {len(h)}"
        assert all(c in "0123456789abcdef" for c in h), "hash 应为小写十六进制"


# ---------------------------------------------------------------------------
# Property 3: content_hash 区分性
# **Validates: Requirements 5.5**
# ---------------------------------------------------------------------------

class TestProperty3ContentHashDistinction:
    """Property 3: content_hash 区分性。

    *For any* 两组不同的 (payload, is_delete) 输入（payload 不同或
    is_delete 不同），_compute_content_hash 应产生不同的哈希值。
    """

    @given(
        payload1=st.dictionaries(st.text(), st.text()),
        is_delete1=st.sampled_from([0, 1]),
        payload2=st.dictionaries(st.text(), st.text()),
        is_delete2=st.sampled_from([0, 1]),
    )
    @settings(max_examples=200, suppress_health_check=[HealthCheck.too_slow])
    def test_property_3_different_inputs_different_hashes(
        self,
        payload1: dict,
        is_delete1: int,
        payload2: dict,
        is_delete2: int,
    ):
        """不同 (payload, is_delete) 对应不同 hash。"""
        from hypothesis import assume

        assume((payload1, is_delete1) != (payload2, is_delete2))

        h1 = BaseOdsTask._compute_content_hash({}, payload=payload1, is_delete=is_delete1)
        h2 = BaseOdsTask._compute_content_hash({}, payload=payload2, is_delete=is_delete2)
        assert h1 != h2, (
            f"区分性违反: 不同输入产生相同 hash {h1!r}\n"
            f"  input1=({payload1!r}, {is_delete1})\n"
            f"  input2=({payload2!r}, {is_delete2})"
        )

    @given(
        payload=st.dictionaries(st.text(), st.text()),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_3_is_delete_flip_changes_hash(self, payload: dict):
        """同一 payload，is_delete 从 0 变 1 应产生不同 hash。"""
        h0 = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=0)
        h1 = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=1)
        assert h0 != h1, f"is_delete 翻转未改变 hash: {h0!r}"


import logging
from hypothesis import assume
from tasks.ods.ods_tasks import ODS_TASK_CLASSES, ODS_TASK_SPECS
from .task_test_utils import (
    create_test_config,
    FakeDBOperations,
    FakeAPIClient,
    FakeCursor,
    FakeConnection,
)

# ---------------------------------------------------------------------------
# 增强版 FakeDB：支持预设 content_hash 返回值（用于 Property 4）
# 复用 test_debug_ods_properties.py 中的 PkAwareFake* 模式
# ---------------------------------------------------------------------------

class _HashAwareFakeCursor(FakeCursor):
    """扩展 FakeCursor，支持 PK 列查询和预设 content_hash 返回。

    当 _fetch_latest_content_hashes 发出 DISTINCT ON 查询时，
    从 _preset_hashes 中返回匹配的 (id, content_hash) 行。
    """

    def __init__(self, recorder, db_ops=None, pk_map=None, columns_map=None, preset_hashes=None):
        super().__init__(recorder, db_ops)
        self._pk_map = pk_map or {}
        self._columns_map = columns_map or {}
        # preset_hashes: {table_name: {(pk_val,): "hash_string"}}
        self._preset_hashes = preset_hashes or {}

    def execute(self, sql, params=None):
        sql_text = sql.decode("utf-8", errors="ignore") if isinstance(sql, (bytes, bytearray)) else str(sql)
        self.recorder.append({"sql": sql_text.strip(), "params": params})
        self._fetchall_rows = []

        lowered = sql_text.lower()

        # 拦截 information_schema.columns 查询
        if "from information_schema.columns" in lowered:
            table_name = None
            if params and len(params) >= 2:
                table_name = params[1]
            if table_name and table_name in self._columns_map:
                self._fetchall_rows = list(self._columns_map[table_name])
            else:
                self._fetchall_rows = self._fake_columns(table_name)
            return

        # 拦截 PK 约束查询
        if "from information_schema.table_constraints" in lowered:
            table_name = None
            if params and len(params) >= 2:
                table_name = params[1]
            if table_name and table_name in self._pk_map:
                self._fetchall_rows = [(col,) for col in self._pk_map[table_name]]
            else:
                self._fetchall_rows = [("id",)]
            return

        # 拦截 _fetch_latest_content_hashes 的 DISTINCT ON 查询
        if "distinct on" in lowered and "content_hash" in lowered:
            # 从 preset_hashes 中查找匹配的表
            for table_name, hash_map in self._preset_hashes.items():
                if table_name in sql_text:
                    # 返回所有预设的 (pk_val, content_hash) 行
                    rows = [(*key, h) for key, h in hash_map.items()]
                    self._fetchall_rows = rows
                    self._pending_rows = []  # 清除 mogrify 积累的行
                    return
            self._fetchall_rows = []
            self._pending_rows = []
            return

        # 默认：处理 INSERT 等语句
        if self._pending_rows:
            self.rowcount = len(self._pending_rows)
            self._record_upserts(sql_text)
            # 若 SQL 含 RETURNING，模拟返回每行的 inserted 标志
            if "returning" in lowered:
                self._fetchall_rows = [(True,)] * len(self._pending_rows)
            self._pending_rows = []
        else:
            self.rowcount = 0


class _HashAwareFakeConnection(FakeConnection):
    def __init__(self, db_ops, pk_map=None, columns_map=None, preset_hashes=None):
        super().__init__(db_ops)
        self._pk_map = pk_map or {}
        self._columns_map = columns_map or {}
        self._preset_hashes = preset_hashes or {}

    def cursor(self):
        return _HashAwareFakeCursor(
            self.statements, self._db_ops,
            pk_map=self._pk_map,
            columns_map=self._columns_map,
            preset_hashes=self._preset_hashes,
        )


class _HashAwareFakeDB(FakeDBOperations):
    """增强版 FakeDB，支持 PK 列和预设 content_hash。"""

    def __init__(self, pk_map=None, columns_map=None, preset_hashes=None):
        super().__init__()
        self._pk_map = pk_map or {}
        self._columns_map = columns_map or {}
        self._preset_hashes = preset_hashes or {}
        self.conn = _HashAwareFakeConnection(
            self, self._pk_map, self._columns_map, self._preset_hashes,
        )


# ---------------------------------------------------------------------------
# 辅助常量和函数
# ---------------------------------------------------------------------------

_PAYMENT_CODE = "ODS_PAYMENT"
_PAYMENT_ENDPOINT = "/PayLog/GetPayLogListPage"
_PAYMENT_TABLE = "payment_transactions"

# 标准 ODS 列定义（含 content_hash + is_delete）
_STANDARD_ODS_COLUMNS = [
    ("id", "bigint", "int8"),
    ("record_index", "integer", "int4"),
    ("content_hash", "text", "text"),
    ("source_file", "text", "text"),
    ("source_endpoint", "text", "text"),
    ("fetched_at", "timestamp with time zone", "timestamptz"),
    ("payload", "jsonb", "jsonb"),
    ("is_delete", "integer", "int4"),
]

_valid_id = st.integers(min_value=1, max_value=2**53)

_ods_record_with_id = st.fixed_dictionaries({
    "id": _valid_id,
    "payAmount": st.text(alphabet="0123456789.", min_size=1, max_size=8),
})


def _build_config(tmp_path, conflict_mode="update"):
    archive_dir = tmp_path / "archive"
    temp_dir = tmp_path / "temp"
    cfg = create_test_config("ONLINE", archive_dir, temp_dir)
    cfg.config.setdefault("run", {})
    cfg.config["run"]["ods_conflict_mode"] = conflict_mode
    cfg.config["run"]["snapshot_missing_delete"] = False
    return cfg


def _get_task_spec(code):
    for spec in ODS_TASK_SPECS:
        if spec.code == code:
            return spec
    raise KeyError(f"未找到任务 spec: {code}")


# ---------------------------------------------------------------------------
# Property 4: skip_unchanged 跳过内容未变的记录
# **Validates: Requirements 4.3, 8.5**
# ---------------------------------------------------------------------------

class TestProperty4SkipUnchanged:
    """Property 4: skip_unchanged 跳过内容未变的记录。

    *For any* ODS 任务（skip_unchanged=True），当一条记录的 content_hash
    与数据库中该业务 ID 最新版本的 content_hash 相同时，该记录应被计入
    skipped 而非 inserted。
    """

    @given(record=_ods_record_with_id)
    @settings(max_examples=100, suppress_health_check=[
        HealthCheck.too_slow, HealthCheck.function_scoped_fixture,
    ])
    def test_property_4_unchanged_record_is_skipped(self, tmp_path, record):
        """内容未变的记录应被跳过（skipped），不应被插入。

        策略：先计算记录的 content_hash（与任务内部逻辑一致），
        然后预设到 FakeDB 中，模拟"数据库中已有相同 hash"的场景。
        """
        spec = _get_task_spec(_PAYMENT_CODE)
        task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

        # snapshot_missing_delete=False → default_is_delete=None
        # → _normalize_is_delete_flag 不会注入 is_delete
        # → payload 保持原样，is_delete 取 merged_rec.get("is_delete", 0) = 0
        expected_hash = BaseOdsTask._compute_content_hash(
            {}, payload=record, is_delete=0,
        )

        rec_id = record["id"]

        preset_hashes = {
            _PAYMENT_TABLE: {(rec_id,): expected_hash},
        }

        config = _build_config(tmp_path)
        db = _HashAwareFakeDB(
            pk_map={_PAYMENT_TABLE: ["id", "content_hash"]},
            columns_map={_PAYMENT_TABLE: _STANDARD_ODS_COLUMNS},
            preset_hashes=preset_hashes,
        )
        api = FakeAPIClient({spec.endpoint: [record]})
        logger = logging.getLogger("prop4")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        # 核心断言：内容未变的记录应被跳过
        assert counts["skipped"] >= 1, (
            f"skip_unchanged=True 且 hash 相同时，记录应被跳过，"
            f"实际 counts={counts}"
        )
        assert counts["inserted"] == 0, (
            f"内容未变的记录不应被插入，实际 inserted={counts['inserted']}"
        )

    @given(record=_ods_record_with_id)
    @settings(max_examples=100, suppress_health_check=[
        HealthCheck.too_slow, HealthCheck.function_scoped_fixture,
    ])
    def test_property_4_changed_record_is_not_skipped(self, tmp_path, record):
        """内容已变的记录不应被跳过（应被插入或更新）。

        策略：预设一个不同的 hash 到 FakeDB，模拟"内容已变"。
        """
        spec = _get_task_spec(_PAYMENT_CODE)
        task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

        rec_id = record["id"]

        # 预设一个完全不同的 hash（全零），确保与实际 hash 不同
        preset_hashes = {
            _PAYMENT_TABLE: {(rec_id,): "0" * 64},
        }

        config = _build_config(tmp_path)
        db = _HashAwareFakeDB(
            pk_map={_PAYMENT_TABLE: ["id", "content_hash"]},
            columns_map={_PAYMENT_TABLE: _STANDARD_ODS_COLUMNS},
            preset_hashes=preset_hashes,
        )
        api = FakeAPIClient({spec.endpoint: [record]})
        logger = logging.getLogger("prop4_changed")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        # 内容已变时，记录应被写入（inserted 或 updated）
        assert counts["inserted"] + counts["updated"] >= 1, (
            f"内容已变的记录应被写入，实际 counts={counts}"
        )


# ---------------------------------------------------------------------------
# Property 5: 记录数闭合不变量
# **Validates: Requirements 8.3**
# ---------------------------------------------------------------------------

class TestProperty5RecordCountClosure:
    """Property 5: 记录数闭合不变量。

    *For any* 非空记录列表被写入 ODS 时，
    fetched == inserted + updated + skipped 恒成立。
    """

    @given(records=st.lists(_ods_record_with_id, min_size=1, max_size=20))
    @settings(max_examples=100, suppress_health_check=[
        HealthCheck.too_slow, HealthCheck.function_scoped_fixture,
    ])
    def test_property_5_record_count_closure(self, tmp_path, records):
        """fetched == inserted + updated + skipped 对任意非空记录列表成立。"""
        spec = _get_task_spec(_PAYMENT_CODE)
        task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

        config = _build_config(tmp_path)
        db = _HashAwareFakeDB(
            pk_map={_PAYMENT_TABLE: ["id", "content_hash"]},
            columns_map={_PAYMENT_TABLE: _STANDARD_ODS_COLUMNS},
        )
        api = FakeAPIClient({spec.endpoint: records})
        logger = logging.getLogger("prop5")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        fetched = counts["fetched"]
        inserted = counts["inserted"]
        updated = counts["updated"]
        skipped = counts["skipped"]

        # 核心不变量
        assert fetched == inserted + updated + skipped, (
            f"记录数不闭合: fetched={fetched}, "
            f"inserted={inserted}, updated={updated}, skipped={skipped}"
        )
        # fetched 应等于 API 提供的记录数
        assert fetched == len(records), (
            f"fetched({fetched}) != len(records)({len(records)})"
        )

    @given(records=st.lists(_ods_record_with_id, min_size=1, max_size=15))
    @settings(max_examples=100, suppress_health_check=[
        HealthCheck.too_slow, HealthCheck.function_scoped_fixture,
    ])
    def test_property_5_closure_with_preset_hashes(self, tmp_path, records):
        """即使部分记录被 skip_unchanged 跳过，闭合不变量仍成立。

        策略：为部分记录预设相同的 hash（模拟已存在），验证闭合。
        """
        spec = _get_task_spec(_PAYMENT_CODE)
        task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

        # 为前半部分记录预计算 hash 并预设到 DB
        half = max(1, len(records) // 2)
        preset = {}
        for rec in records[:half]:
            # snapshot_missing_delete=False → payload 不含 is_delete
            h = BaseOdsTask._compute_content_hash(
                {}, payload=rec, is_delete=0,
            )
            preset[(rec["id"],)] = h

        config = _build_config(tmp_path)
        db = _HashAwareFakeDB(
            pk_map={_PAYMENT_TABLE: ["id", "content_hash"]},
            columns_map={_PAYMENT_TABLE: _STANDARD_ODS_COLUMNS},
            preset_hashes={_PAYMENT_TABLE: preset},
        )
        api = FakeAPIClient({spec.endpoint: records})
        logger = logging.getLogger("prop5_mixed")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        fetched = counts["fetched"]
        inserted = counts["inserted"]
        updated = counts["updated"]
        skipped = counts["skipped"]

        # 核心不变量：无论有多少记录被跳过，闭合必须成立
        assert fetched == inserted + updated + skipped, (
            f"记录数不闭合: fetched={fetched}, "
            f"inserted={inserted}, updated={updated}, skipped={skipped}"
        )


# ---------------------------------------------------------------------------
# 软删除属性测试 — Property 6, 7, 8
# 通过 patch execute_values 绕过 psycopg2 mogrify 兼容性问题
# ---------------------------------------------------------------------------

import json
import hashlib
import re as _re
import tempfile
from datetime import datetime, timezone
from unittest.mock import patch

from tasks.ods.ods_tasks import SnapshotMode


# ---------------------------------------------------------------------------
# _SoftDeleteFakeCursor: 拦截 _mark_missing_as_deleted 内部的 SQL
# execute_values 调用通过 patch 转为直接 cursor.execute，无需 mogrify
# ---------------------------------------------------------------------------

class _SoftDeleteFakeCursor:
    """专为 _mark_missing_as_deleted 测试设计的 FakeCursor。"""

    def __init__(self, *, columns_map, pk_map, existing_ids, latest_rows):
        self._columns_map = columns_map
        self._pk_map = pk_map
        self._existing_ids = existing_ids
        self._latest_rows = latest_rows
        self.all_sql: list[dict] = []
        self.insert_sqls: list[dict] = []
        self.update_sqls: list[dict] = []
        self._fetchall_rows: list = []
        self.rowcount = 0
        self.connection = type("_Conn", (), {"encoding": "UTF8"})()
        self._latest_query_consumed = False

    def execute(self, sql, params=None):
        sql_text = sql.decode("utf-8", errors="ignore") if isinstance(sql, (bytes, bytearray)) else str(sql)
        self.all_sql.append({"sql": sql_text.strip(), "params": params})
        self._fetchall_rows = []
        self.rowcount = 0
        lowered = sql_text.lower()

        if "from information_schema.columns" in lowered:
            tbl = params[1] if params and len(params) >= 2 else None
            if tbl and tbl in self._columns_map:
                self._fetchall_rows = list(self._columns_map[tbl])
            return

        if "from information_schema.table_constraints" in lowered:
            tbl = params[1] if params and len(params) >= 2 else None
            if tbl and tbl in self._pk_map:
                self._fetchall_rows = [(c,) for c in self._pk_map[tbl]]
            return

        # 快照范围查询（步骤 1）
        if "select distinct" in lowered and "is_delete" in lowered and "with" not in lowered:
            self._fetchall_rows = [tuple(k) for k in self._existing_ids]
            return

        # execute_values 的 WITH keys... DISTINCT ON... 查询（步骤 3）
        if "with keys" in lowered and "distinct on" in lowered and not self._latest_query_consumed:
            self._fetchall_rows = list(self._latest_rows)
            self._latest_query_consumed = True
            return

        # INSERT（步骤 5）
        if "insert into" in lowered:
            self.insert_sqls.append({"sql": sql_text.strip(), "params": params})
            val_part = sql_text.split("VALUES", 1)[-1] if "VALUES" in sql_text.upper() else ""
            self.rowcount = max(1, len(_re.findall(r"\([^)]+\)", val_part)))
            return

        # UPDATE（不应出现）
        if lowered.lstrip().startswith("update "):
            self.update_sqls.append({"sql": sql_text.strip(), "params": params})
            return

    def fetchall(self):
        return list(self._fetchall_rows)

    def fetchone(self):
        return self._fetchall_rows[0] if self._fetchall_rows else None

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return False


class _SoftDeleteFakeConnection:
    def __init__(self, cursor_instance):
        self._cursor = cursor_instance
        self.statements = []

    def cursor(self):
        return self._cursor


class _SoftDeleteFakeDB:
    """专为软删除测试设计的 FakeDB。"""

    def __init__(self, *, columns_map, pk_map, existing_ids, latest_rows):
        self.cursor_instance = _SoftDeleteFakeCursor(
            columns_map=columns_map, pk_map=pk_map,
            existing_ids=existing_ids, latest_rows=latest_rows,
        )
        self.conn = _SoftDeleteFakeConnection(self.cursor_instance)
        self.commits = 0
        self.rollbacks = 0

    def commit(self):
        self.commits += 1

    def rollback(self):
        self.rollbacks += 1


# ---------------------------------------------------------------------------
# 辅助函数
# ---------------------------------------------------------------------------

_SOFT_DELETE_COLUMNS = [
    ("id", "bigint", "int8"),
    ("content_hash", "text", "text"),
    ("source_file", "text", "text"),
    ("source_endpoint", "text", "text"),
    ("fetched_at", "timestamp with time zone", "timestamptz"),
    ("payload", "jsonb", "jsonb"),
    ("is_delete", "integer", "int4"),
]

_SD_TABLE = "ods.test_soft_delete"
_SD_TABLE_SHORT = "test_soft_delete"


def _build_latest_row(*, biz_id, payload, is_delete=0, content_hash=None):
    """构建模拟的最新版本行 tuple，列顺序与 _SOFT_DELETE_COLUMNS 一致。"""
    if content_hash is None:
        payload_str = json.dumps(
            payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
        )
        content_hash = hashlib.sha256(
            f"{payload_str}|{is_delete}".encode("utf-8")
        ).hexdigest()
    return (
        biz_id,
        content_hash,
        "test_source",
        "/test/endpoint",
        datetime(2025, 1, 1, tzinfo=timezone.utc),
        payload,
        is_delete,
    )


def _fake_execute_values(cur, sql, argslist, page_size=100):
    """替代 psycopg2.extras.execute_values，直接调用 cur.execute。

    将 argslist 中的每个 tuple 格式化为 SQL VALUES 子句，
    拼接到 sql 模板的 %s 位置，然后调用 cur.execute。
    """
    if not argslist:
        return
    # 将每个 tuple 转为字符串表示
    parts = []
    for row in argslist:
        vals = []
        for v in row:
            if v is None:
                vals.append("NULL")
            elif isinstance(v, str):
                escaped = v.replace("'", "''")
                vals.append(f"'{escaped}'")
            elif isinstance(v, (dict, list)):
                j = json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
                escaped = j.replace("'", "''")
                vals.append(f"'{escaped}'")
            elif hasattr(v, "adapted"):
                # psycopg2 Json wrapper
                j = json.dumps(v.adapted, ensure_ascii=False)
                escaped = j.replace("'", "''")
                vals.append(f"'{escaped}'")
            else:
                vals.append(str(v))
        parts.append(f"({','.join(vals)})")
    values_str = ",".join(parts)
    final_sql = sql.replace("%s", values_str)
    cur.execute(final_sql)


def _make_task_for_soft_delete(db, *, snapshot_mode=SnapshotMode.FULL_TABLE):
    """构造最小化 BaseOdsTask 实例，用于调用 _mark_missing_as_deleted。"""
    spec = OdsTaskSpec(
        code="TEST_SOFT_DELETE",
        class_name="TestSoftDelete",
        table_name=_SD_TABLE,
        endpoint="/test/soft_delete",
        snapshot_mode=snapshot_mode,
        snapshot_time_column=None if snapshot_mode != SnapshotMode.WINDOW else "create_time",
    )
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        config = create_test_config("ONLINE", tmp_path / "archive", tmp_path / "temp")
    task_cls = type("_TestSoftDeleteTask", (BaseOdsTask,), {"SPEC": spec})
    api = FakeAPIClient({"/test/soft_delete": []})
    task = task_cls(config, db, api, logging.getLogger("soft_delete_test"))
    task._table_columns_cache = {}
    task._table_pk_cache = {}
    return task


def _run_mark_missing(task, *, key_values=None, allow_empty=True):
    """在 patch execute_values 的上下文中调用 _mark_missing_as_deleted。"""
    with patch("tasks.ods.ods_tasks.execute_values", _fake_execute_values):
        return task._mark_missing_as_deleted(
            table=_SD_TABLE,
            business_pk_cols=["id"],
            snapshot_mode=SnapshotMode.FULL_TABLE,
            snapshot_time_column=None,
            window_start=datetime(2025, 1, 1, tzinfo=timezone.utc),
            window_end=datetime(2025, 1, 2, tzinfo=timezone.utc),
            key_values=key_values if key_values is not None else set(),
            allow_empty=allow_empty,
        )


# ---------------------------------------------------------------------------
# Property 6: 软删除构造正确性
# **Validates: Requirements 7.1, 7.2, 7.4**
# ---------------------------------------------------------------------------

class TestProperty6SoftDeleteConstruction:
    """Property 6: 软删除构造正确性。

    *For any* 快照对比中发现的缺失业务 ID，_mark_missing_as_deleted 应读取
    该 ID 的最新版本行，构造一条 is_delete=1 的新版本行，其 content_hash
    基于原始 payload + is_delete=1 重算，并通过 INSERT（而非 UPDATE）写入。
    """

    @given(
        biz_id=st.integers(min_value=1, max_value=2**53),
        payload=st.fixed_dictionaries({
            "name": st.text(min_size=1, max_size=20),
            "value": st.text(min_size=1, max_size=10),
        }),
    )
    @settings(max_examples=100, deadline=None, suppress_health_check=[HealthCheck.too_slow])
    def test_property_6_soft_delete_inserts_new_version(self, biz_id, payload):
        """缺失 ID 应通过 INSERT 写入 is_delete=1 的新版本行，而非 UPDATE。"""
        latest_row = _build_latest_row(biz_id=biz_id, payload=payload, is_delete=0)
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids={(biz_id,)},
            latest_rows=[latest_row],
        )
        task = _make_task_for_soft_delete(db)
        deleted = _run_mark_missing(task)

        cur = db.cursor_instance
        # 核心断言 1: 应有 INSERT 语句
        assert len(cur.insert_sqls) >= 1, (
            f"应通过 INSERT 写入删除版本行，实际无 INSERT。"
            f"所有 SQL: {[s['sql'][:80] for s in cur.all_sql]}"
        )
        # 核心断言 2: 不应有 UPDATE 语句
        assert len(cur.update_sqls) == 0, (
            f"软删除不应使用 UPDATE: {[s['sql'][:80] for s in cur.update_sqls]}"
        )
        # 核心断言 3: 返回值 >= 1
        assert deleted >= 1, f"应返回 >= 1，实际 {deleted}"

    @given(
        biz_id=st.integers(min_value=1, max_value=2**53),
        payload=st.fixed_dictionaries({
            "name": st.text(min_size=1, max_size=20),
            "value": st.text(min_size=1, max_size=10),
        }),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_6_content_hash_recomputed_with_is_delete_1(self, biz_id, payload):
        """删除版本行的 content_hash 应基于 payload + is_delete=1 重算。"""
        original_hash = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=0)
        expected_delete_hash = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=1)
        latest_row = _build_latest_row(
            biz_id=biz_id, payload=payload, is_delete=0, content_hash=original_hash,
        )
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids={(biz_id,)},
            latest_rows=[latest_row],
        )
        task = _make_task_for_soft_delete(db)
        _run_mark_missing(task)

        cur = db.cursor_instance
        assert len(cur.insert_sqls) >= 1, "应有 INSERT SQL"
        insert_sql = cur.insert_sqls[0]["sql"]
        # INSERT SQL 中应包含重算后的 content_hash
        assert expected_delete_hash in insert_sql, (
            f"INSERT 应含 hash {expected_delete_hash[:16]}...，"
            f"原始 hash={original_hash[:16]}...，SQL: {insert_sql[:200]}"
        )
        # is_delete 从 0→1 后 hash 必须不同
        assert expected_delete_hash != original_hash



# ---------------------------------------------------------------------------
# Property 7: 软删除幂等性
# **Validates: Requirements 7.3, 8.7**
# ---------------------------------------------------------------------------

class TestProperty7SoftDeleteIdempotency:
    """Property 7: 软删除幂等性。

    *For any* 业务 ID，若其最新版本已经是 is_delete=1，再次执行
    _mark_missing_as_deleted 不应插入新的删除版本行。
    """

    @given(
        biz_id=st.integers(min_value=1, max_value=2**53),
        payload=st.fixed_dictionaries({
            "name": st.text(min_size=1, max_size=20),
            "value": st.text(min_size=1, max_size=10),
        }),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_7_already_deleted_no_new_insert(self, biz_id, payload):
        """最新版本已是 is_delete=1 时，不应插入新的删除版本行。"""
        delete_hash = BaseOdsTask._compute_content_hash({}, payload=payload, is_delete=1)
        latest_row = _build_latest_row(
            biz_id=biz_id, payload=payload, is_delete=1, content_hash=delete_hash,
        )
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids={(biz_id,)},
            latest_rows=[latest_row],
        )
        task = _make_task_for_soft_delete(db)
        deleted = _run_mark_missing(task)

        cur = db.cursor_instance
        assert deleted == 0, (
            f"最新版本已是 is_delete=1 时不应插入新行，实际 deleted={deleted}"
        )
        assert len(cur.insert_sqls) == 0, (
            f"不应有 INSERT: {[s['sql'][:80] for s in cur.insert_sqls]}"
        )

    @given(
        biz_ids=st.lists(
            st.integers(min_value=1, max_value=10000),
            min_size=2, max_size=8, unique=True,
        ),
    )
    @settings(max_examples=50, suppress_health_check=[HealthCheck.too_slow])
    def test_property_7_mixed_deleted_and_active(self, biz_ids):
        """混合场景：部分 ID 已删除、部分未删除，仅未删除的应产生 INSERT。"""
        half = max(1, len(biz_ids) // 2)
        already_deleted_ids = biz_ids[:half]
        active_ids = biz_ids[half:]

        latest_rows = []
        for bid in already_deleted_ids:
            p = {"id": bid, "status": "deleted"}
            latest_rows.append(_build_latest_row(biz_id=bid, payload=p, is_delete=1))
        for bid in active_ids:
            p = {"id": bid, "status": "active"}
            latest_rows.append(_build_latest_row(biz_id=bid, payload=p, is_delete=0))

        all_ids = {(bid,) for bid in biz_ids}
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids=all_ids,
            latest_rows=latest_rows,
        )
        task = _make_task_for_soft_delete(db)
        _run_mark_missing(task)

        cur = db.cursor_instance
        # 有活跃 ID 时应有 INSERT
        if active_ids:
            assert len(cur.insert_sqls) >= 1, (
                f"有 {len(active_ids)} 个活跃 ID 应被标记删除，但无 INSERT"
            )
        # 不应有 UPDATE
        assert len(cur.update_sqls) == 0, "不应有 UPDATE SQL"


# ---------------------------------------------------------------------------
# Property 8: 软删除不修改历史版本
# **Validates: Requirements 7.4, 8.6**
# ---------------------------------------------------------------------------

class TestProperty8SoftDeleteNoHistoryModification:
    """Property 8: 软删除不修改历史版本。

    *For any* 软删除操作执行后，数据库中该业务 ID 的所有历史版本行
    （执行前已存在的行）的内容应保持不变——不应有 UPDATE 语句作用于 ODS 表。
    """

    @given(
        biz_id=st.integers(min_value=1, max_value=2**53),
        payload=st.fixed_dictionaries({
            "name": st.text(min_size=1, max_size=20),
            "value": st.text(min_size=1, max_size=10),
        }),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_8_no_update_on_ods_table(self, biz_id, payload):
        """软删除操作不应产生任何 UPDATE 语句。"""
        latest_row = _build_latest_row(biz_id=biz_id, payload=payload, is_delete=0)
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids={(biz_id,)},
            latest_rows=[latest_row],
        )
        task = _make_task_for_soft_delete(db)
        _run_mark_missing(task)

        cur = db.cursor_instance
        # 核心断言: 无 UPDATE
        assert len(cur.update_sqls) == 0, (
            f"软删除不应产生 UPDATE: {[s['sql'][:120] for s in cur.update_sqls]}"
        )
        # 额外验证：所有 SQL 中不含 UPDATE + 表名
        for stmt in cur.all_sql:
            sql_lower = stmt["sql"].lower()
            assert not (
                "update" in sql_lower and _SD_TABLE.lower() in sql_lower
            ), f"不应有 UPDATE 作用于 {_SD_TABLE}，发现: {stmt['sql'][:120]}"

    @given(
        biz_ids=st.lists(
            st.integers(min_value=1, max_value=10000),
            min_size=1, max_size=10, unique=True,
        ),
    )
    @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow])
    def test_property_8_multiple_ids_no_update(self, biz_ids):
        """多个缺失 ID 的软删除操作也不应产生 UPDATE。"""
        latest_rows = []
        for bid in biz_ids:
            p = {"id": bid, "data": "test"}
            latest_rows.append(_build_latest_row(biz_id=bid, payload=p, is_delete=0))

        all_ids = {(bid,) for bid in biz_ids}
        db = _SoftDeleteFakeDB(
            columns_map={_SD_TABLE_SHORT: _SOFT_DELETE_COLUMNS},
            pk_map={_SD_TABLE_SHORT: ["id", "content_hash"]},
            existing_ids=all_ids,
            latest_rows=latest_rows,
        )
        task = _make_task_for_soft_delete(db)
        _run_mark_missing(task)

        cur = db.cursor_instance
        assert len(cur.update_sqls) == 0, (
            f"多 ID 软删除不应产生 UPDATE: {[s['sql'][:120] for s in cur.update_sqls]}"
        )
        if biz_ids:
            assert len(cur.insert_sqls) >= 1, "应有 INSERT 写入删除版本行"
