# -*- coding: utf-8 -*-
"""ODS 层属性测试 — 验证 BaseOdsTask 的核心正确性属性。

Feature: etl-flow-debug
使用 hypothesis 对 ODS 任务的关键行为进行属性测试。
"""
from __future__ import annotations

import logging
import os
import re
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest
from hypothesis import given, settings, assume, 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 BaseOdsTask, ODS_TASK_CLASSES, ODS_TASK_SPECS, OdsTaskSpec
from .task_test_utils import (
    create_test_config, FakeDBOperations, FakeAPIClient, FakeCursor, FakeConnection,
)

# ---------------------------------------------------------------------------
# 增强版 FakeCursor：返回 PK 列信息，使冲突处理和主键跳过逻辑生效
# ---------------------------------------------------------------------------

class PkAwareFakeCursor(FakeCursor):
    """扩展 FakeCursor，对 table_constraints 查询返回指定的 PK 列。

    通过实例级 _pk_map / _columns_map 配置（由 PkAwareFakeDB 注入），
    避免类级别状态在并发测试间互相干扰。
    """

    def __init__(self, recorder, db_ops=None, pk_map=None, columns_map=None):
        super().__init__(recorder, db_ops)
        self._pk_map = pk_map or {}
        self._columns_map = columns_map 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()
        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
        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:
                # 默认返回 id 作为 PK
                self._fetchall_rows = [("id",)]
            return

        if self._pending_rows:
            self.rowcount = len(self._pending_rows)
            self._record_upserts(sql_text)
            self._pending_rows = []
        else:
            self.rowcount = 0


class PkAwareFakeConnection(FakeConnection):
    """使用 PkAwareFakeCursor 的连接。"""

    def __init__(self, db_ops, pk_map=None, columns_map=None):
        super().__init__(db_ops)
        self._pk_map = pk_map or {}
        self._columns_map = columns_map or {}

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


class PkAwareFakeDB(FakeDBOperations):
    """增强版 FakeDB，返回 PK 列信息。"""

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


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

def _build_config(tmp_path: Path, conflict_mode: str = "update",
                  snapshot_missing_delete: bool = False) -> "AppConfig":
    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"] = snapshot_missing_delete
    return cfg


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


# ---------------------------------------------------------------------------
# hypothesis 策略
# 使用 id 字段名（与 DB PK 列名一致），确保记录不会因字段名不匹配被跳过
# ---------------------------------------------------------------------------
_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),
})

_missing_pk_record = st.fixed_dictionaries({
    "id": st.sampled_from([None, ""]),
    "payAmount": st.just("0.00"),
})

# ODS_PAYMENT 的 endpoint 和 table
_PAYMENT_CODE = "ODS_PAYMENT"
_PAYMENT_ENDPOINT = "/PayLog/GetPayLogListPage"
_PAYMENT_TABLE = "payment_transactions"

# ODS_ASSISTANT_ACCOUNT 的 endpoint 和 table
_ASSISTANT_CODE = "ODS_ASSISTANT_ACCOUNT"
_ASSISTANT_TABLE = "assistant_accounts_master"

# 包含 is_delete 列的列定义（用于 Property 5）
_COLUMNS_WITH_IS_DELETE = [
    ("id", "bigint", "int8"),
    ("sitegoodsstockid", "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"),
]


# ===========================================================================
# Property 1: ODS 任务提取记录数一致性
# Feature: etl-flow-debug, Property 1: ODS 任务提取记录数一致性
# Validates: Requirements 1.1, 1.2
# ===========================================================================

@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(records=st.lists(_ods_record_with_id, min_size=1, max_size=20))
def test_property1_ods_record_count_consistency(tmp_path, records):
    """对任意非空记录列表，fetched == inserted + updated + skipped。

    **Validates: Requirements 1.1, 1.2**
    """
    spec = _get_task_spec(_PAYMENT_CODE)
    task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

    config = _build_config(tmp_path)
    db = PkAwareFakeDB()
    api = FakeAPIClient({spec.endpoint: records})
    logger = logging.getLogger("prop1")

    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"]

    # 核心属性：fetched == inserted + updated + 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)})"
    )


# ===========================================================================
# Property 2: ODS 冲突处理策略正确性
# Feature: etl-flow-debug, Property 2: ODS 冲突处理策略正确性
# Validates: Requirements 1.3
# ===========================================================================

@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(conflict_mode=st.sampled_from(["nothing", "backfill", "update"]))
def test_property2_ods_conflict_mode_sql(tmp_path, conflict_mode):
    """对任意 ods_conflict_mode，生成的 SQL 应包含对应的冲突处理子句。

    **Validates: Requirements 1.3**
    """
    spec = _get_task_spec(_PAYMENT_CODE)
    task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

    config = _build_config(tmp_path, conflict_mode=conflict_mode)
    db = PkAwareFakeDB()
    # 使用 id 字段确保记录不被跳过
    sample = [{"id": 1001, "payAmount": "50.00"}]
    api = FakeAPIClient({spec.endpoint: sample})
    logger = logging.getLogger("prop2")

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

    # INSERT SQL 通过 execute_values → cur.execute 记录在 db.conn.statements 中
    all_sql = [stmt["sql"] for stmt in db.conn.statements]
    insert_sqls = [s for s in all_sql if "INSERT" in s.upper() and "VALUES" in s.upper()]
    assert insert_sqls, (
        f"应至少有一条 INSERT...VALUES 语句，"
        f"实际 SQL: {[s[:100] for s in all_sql]}, counts={result['counts']}"
    )
    sql_upper = insert_sqls[0].upper()

    if conflict_mode == "nothing":
        assert "DO NOTHING" in sql_upper, (
            f"conflict_mode=nothing 时 SQL 应包含 DO NOTHING"
        )
    elif conflict_mode == "backfill":
        assert "COALESCE" in sql_upper, (
            f"conflict_mode=backfill 时 SQL 应包含 COALESCE"
        )
    elif conflict_mode == "update":
        assert "IS DISTINCT FROM" in sql_upper, (
            f"conflict_mode=update 时 SQL 应包含 IS DISTINCT FROM"
        )


# ===========================================================================
# Property 3: ODS 跳过缺失主键记录
# Feature: etl-flow-debug, Property 3: ODS 跳过缺失主键记录
# Validates: Requirements 1.4
# ===========================================================================

@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(
    valid_records=st.lists(_ods_record_with_id, min_size=0, max_size=10),
    missing_pk_records=st.lists(_missing_pk_record, min_size=1, max_size=10),
)
def test_property3_ods_skip_missing_pk(tmp_path, valid_records, missing_pk_records):
    """对任意包含缺失主键的记录集合，skipped 应 >= 缺失主键的记录数。

    **Validates: Requirements 1.4**
    """
    spec = _get_task_spec(_PAYMENT_CODE)
    task_cls = ODS_TASK_CLASSES[_PAYMENT_CODE]

    all_records = valid_records + missing_pk_records
    assume(len(all_records) > 0)

    config = _build_config(tmp_path)
    db = PkAwareFakeDB()
    api = FakeAPIClient({spec.endpoint: all_records})
    logger = logging.getLogger("prop3")

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

    counts = result["counts"]
    skipped = counts["skipped"]

    # 缺失主键的记录至少应被跳过
    assert skipped >= len(missing_pk_records), (
        f"skipped({skipped}) < missing_pk_count({len(missing_pk_records)})"
    )

    # 写入的行不应包含缺失主键的记录
    if db.upserts:
        for upsert in db.upserts:
            for row in upsert.get("rows", []):
                pk_val = row.get("id")
                assert pk_val is not None and pk_val != "", (
                    f"写入行中出现缺失主键: {row}"
                )


# ===========================================================================
# Property 4: ODS content_hash 去重
# Feature: etl-flow-debug, Property 4: ODS content_hash 去重
# Validates: Requirements 1.5
# ===========================================================================

@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(record=_ods_record_with_id)
def test_property4_ods_content_hash_deterministic(tmp_path, record):
    """对任意记录，相同 payload + is_delete 组合的 content_hash 应确定性相同。

    **Validates: Requirements 1.5, 5.1, 5.4**

    新签名：_compute_content_hash(record, payload=..., is_delete=...)
    hash 基于原始 payload（未展平）+ is_delete 计算，fetched_at 天然不参与。
    """
    merged = BaseOdsTask._merge_record_layers(record)
    is_delete = 0

    hash1 = BaseOdsTask._compute_content_hash(merged, payload=record, is_delete=is_delete)
    hash2 = BaseOdsTask._compute_content_hash(merged, payload=record, is_delete=is_delete)
    assert hash1 == hash2, "同一 payload + is_delete 的 content_hash 应确定性相同"

    # fetched_at 不影响 hash（payload 天然不含 fetched_at）
    merged_with_ts1 = {**merged, "fetched_at": "2025-01-01T10:00:00+08:00"}
    merged_with_ts2 = {**merged, "fetched_at": "2025-06-15T23:59:59+08:00"}

    hash3 = BaseOdsTask._compute_content_hash(merged_with_ts1, payload=record, is_delete=is_delete)
    hash4 = BaseOdsTask._compute_content_hash(merged_with_ts2, payload=record, is_delete=is_delete)
    assert hash3 == hash4, "merged_rec 中 fetched_at 不同不影响 hash（基于 payload 计算）"
    assert hash1 == hash3, "hash 仅取决于 payload + is_delete，与 merged_rec 内容无关"


# ===========================================================================
# Property 5: ODS 快照删除标记（INSERT 语义）
# Feature: etl-flow-debug, Property 5: ODS 快照删除标记
# Validates: Requirements 1.7, 7.1, 7.4
# ===========================================================================

@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(
    existing_ids=st.lists(
        st.integers(min_value=1, max_value=10000), min_size=2, max_size=10, unique=True
    ),
)
def test_property5_ods_snapshot_delete_marking(tmp_path, existing_ids):
    """对启用 snapshot_missing_delete 的任务，API 返回真子集时差集应被标记删除。

    **Validates: Requirements 1.7, 7.1, 7.4**

    测试策略：使用 ODS_ASSISTANT_ACCOUNT（snapshot_mode=FULL_TABLE），
    API 只返回 existing_ids 的前半部分，验证 SQL 中包含 INSERT（而非独立 UPDATE）
    来写入 is_delete=1 的删除版本行。

    方案 4 改造后，软删除从 UPDATE 改为 INSERT 删除版本行：
    - _mark_missing_as_deleted 读取缺失 ID 的最新版本行
    - 构造 is_delete=1 的新版本行并 INSERT
    - 不应有独立的 UPDATE ... SET is_delete=1 语句
    """
    if _ASSISTANT_CODE not in ODS_TASK_CLASSES:
        pytest.skip(f"{_ASSISTANT_CODE} 不在 ODS_TASK_CLASSES 中")

    spec = _get_task_spec(_ASSISTANT_CODE)
    task_cls = ODS_TASK_CLASSES[_ASSISTANT_CODE]

    # API 只返回前半部分 ID（真子集）
    subset_size = max(1, len(existing_ids) // 2)
    api_ids = existing_ids[:subset_size]
    assume(len(api_ids) < len(existing_ids))

    api_records = [
        {"id": eid, "assistant_no": f"A{eid}", "nickname": f"助教{eid}"}
        for eid in api_ids
    ]

    config = _build_config(tmp_path, snapshot_missing_delete=True)

    # 配置增强版 FakeDB，使其返回包含 is_delete 列的列信息
    db = PkAwareFakeDB(
        pk_map={_ASSISTANT_TABLE: ["id"]},
        columns_map={_ASSISTANT_TABLE: _COLUMNS_WITH_IS_DELETE},
    )
    api = FakeAPIClient({spec.endpoint: api_records})
    logger = logging.getLogger("prop5")

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

    assert result["status"] == "SUCCESS"

    # 方案 4 后，_mark_missing_as_deleted 通过 INSERT 写入删除版本行
    # 正常写入路径的 INSERT ... ON CONFLICT DO UPDATE 中会包含 "update" 关键字，
    # 但不应有独立的 "UPDATE ods.xxx SET is_delete=1" 语句
    all_sql = [stmt["sql"] for stmt in db.conn.statements]

    # 验证：不应有独立的 UPDATE ... SET is_delete 语句（旧语义）
    has_standalone_update = any(
        sql.lower().lstrip().startswith("update ")
        and "is_delete" in sql.lower()
        for sql in all_sql
    )
    assert not has_standalone_update, (
        f"方案 4 后不应有独立的 UPDATE ... SET is_delete 语句，"
        f"实际 SQL: {[s[:120] for s in all_sql if 'update' in s.lower()]}"
    )

    # 验证：应有 INSERT 语句（正常写入路径 + 可能的软删除 INSERT）
    has_insert = any(
        "insert" in sql.lower() and "values" in sql.lower()
        for sql in all_sql
    )
    assert has_insert, (
        f"应至少有一条 INSERT 语句，"
        f"实际 SQL: {[s[:120] for s in all_sql]}"
    )
