# -*- coding: utf-8 -*-
"""DWD/DWS 层属性测试 — 验证 DwdLoadTask 和 BaseTask 的核心正确性属性。

Feature: etl-flow-debug
使用 hypothesis 对 DWD 列映射完整性、only_tables 过滤和 DWS 分段累加进行属性测试。
"""
from __future__ import annotations

import os
import re
import sys
from pathlib import Path

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.dwd.dwd_load_task import DwdLoadTask
from tasks.base_task import BaseTask


# ---------------------------------------------------------------------------
# Property 6 辅助：判断 ods_expr 是否为"简单列名"
# 排除 JSON 路径表达式、CASE WHEN、函数调用、CAST 等复杂表达式
# ---------------------------------------------------------------------------

def _is_simple_column_name(expr: str) -> bool:
    """判断 ods_expr 是否为简单列名（非表达式）。

    排除条件：包含 ->、CASE、空格、括号、::、单引号等。
    """
    if not expr:
        return False
    if any(ch in expr for ch in ("->", "'", "(", ")", "::", " ")):
        return False
    # 纯标识符：字母/数字/下划线
    return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", expr))


# ---------------------------------------------------------------------------
# Property 6 辅助：收集所有 ODS 源表名（从 TABLE_MAP 的值去重）
# ---------------------------------------------------------------------------

def _get_ods_table_names() -> set[str]:
    """返回 TABLE_MAP 中所有 ODS 源表的 table_name（不含 schema）。"""
    names = set()
    for ods_full in DwdLoadTask.TABLE_MAP.values():
        # "ods.payment_transactions" -> "payment_transactions"
        names.add(ods_full.split(".")[-1])
    return names


# ===========================================================================
# Property 6: DWD FACT_MAPPINGS 列映射完整性
# Feature: etl-flow-debug, Property 6: DWD FACT_MAPPINGS 列映射完整性
# Validates: Requirements 2.4
#
# 验证策略：静态检查 FACT_MAPPINGS 中每个映射条目，当 ods_expr 是简单列名时，
# 该列应存在于对应 ODS 源表的真实列定义中。由于无法在纯单元测试中查询
# information_schema，我们通过 hypothesis 从 FACT_MAPPINGS 中随机抽取条目，
# 验证简单列名的格式合法性和映射关系的内部一致性。
# ===========================================================================

# 构建所有 (dwd_table, dwd_col, ods_expr, cast_type) 的扁平列表
_ALL_MAPPING_ENTRIES: list[tuple[str, str, str, str | None]] = []
for _dwd_tbl, _entries in DwdLoadTask.FACT_MAPPINGS.items():
    for _dwd_col, _ods_expr, _cast_type in _entries:
        _ALL_MAPPING_ENTRIES.append((_dwd_tbl, _dwd_col, _ods_expr, _cast_type))

# 仅保留简单列名的条目（可直接验证是否存在于 ODS 表）
_SIMPLE_COL_ENTRIES: list[tuple[str, str, str, str | None]] = [
    (tbl, col, expr, cast) for tbl, col, expr, cast in _ALL_MAPPING_ENTRIES
    if _is_simple_column_name(expr)
]


@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(idx=st.integers(min_value=0, max_value=max(len(_ALL_MAPPING_ENTRIES) - 1, 0)))
def test_property6_fact_mappings_column_integrity(idx):
    """对任意 FACT_MAPPINGS 条目，验证映射关系的内部一致性。

    **Validates: Requirements 2.4**

    属性：
    1. 每个 dwd_table 必须存在于 TABLE_MAP 中（映射有对应的 ODS 源表）
    2. dwd_col 必须是合法标识符
    3. 当 ods_expr 是简单列名时，它必须是合法标识符
    4. cast_type 为 None 或非空字符串
    """
    assume(len(_ALL_MAPPING_ENTRIES) > 0)
    entry = _ALL_MAPPING_ENTRIES[idx % len(_ALL_MAPPING_ENTRIES)]
    dwd_table, dwd_col, ods_expr, cast_type = entry

    # 1. dwd_table 必须在 TABLE_MAP 中有对应的 ODS 源表
    assert dwd_table in DwdLoadTask.TABLE_MAP, (
        f"FACT_MAPPINGS 中的 {dwd_table} 不在 TABLE_MAP 中"
    )

    # 2. dwd_col 必须是合法标识符
    assert re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", dwd_col), (
        f"dwd_col '{dwd_col}' 不是合法标识符"
    )

    # 3. 当 ods_expr 是简单列名时，验证格式合法
    if _is_simple_column_name(ods_expr):
        assert re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", ods_expr), (
            f"简单列名 '{ods_expr}' 格式不合法"
        )

    # 4. cast_type 为 None 或非空字符串
    if cast_type is not None:
        assert isinstance(cast_type, str) and len(cast_type) > 0, (
            f"cast_type 应为 None 或非空字符串，实际: {cast_type!r}"
        )


# ===========================================================================
# Property 7: DWD only_tables 过滤
# Feature: etl-flow-debug, Property 7: DWD only_tables 过滤
# Validates: Requirements 2.6
#
# 验证策略：模拟 DwdLoadTask.load() 中的 only_tables 过滤逻辑，
# 对任意非空的 only_tables 配置列表，验证过滤后的表集合等于
# 配置列表与 TABLE_MAP 键集合的交集。
# ===========================================================================

_TABLE_MAP_KEYS = list(DwdLoadTask.TABLE_MAP.keys())
# 同时包含不带 schema 的表名，用于生成测试数据
_TABLE_MAP_BASE_NAMES = list({k.split(".")[-1] for k in _TABLE_MAP_KEYS})

# 策略：从 TABLE_MAP 键中随机选取子集作为 only_tables 配置
_only_tables_strategy = st.lists(
    st.one_of(
        # 带 schema 的全名（如 "dwd.dim_site"）
        st.sampled_from(_TABLE_MAP_KEYS),
        # 不带 schema 的表名（如 "dim_site"）
        st.sampled_from(_TABLE_MAP_BASE_NAMES),
        # 不存在的表名（应被过滤掉）
        st.text(alphabet="abcdefghijklmnopqrstuvwxyz_", min_size=3, max_size=15),
    ),
    min_size=1,
    max_size=10,
)


@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(only_tables_cfg=_only_tables_strategy)
def test_property7_dwd_only_tables_filter(only_tables_cfg):
    """对任意非空 only_tables 配置，过滤后的表集合应是配置与 TABLE_MAP 的交集。

    **Validates: Requirements 2.6**

    复现 DwdLoadTask.load() 中的过滤逻辑：
    - 配置列表转为小写集合
    - 遍历 TABLE_MAP，保留 dwd_table.lower() 或 _table_base(dwd_table).lower() 在集合中的表
    """
    # 复现 load() 中的 only_tables 构建逻辑
    only_tables = {
        str(t).strip().lower()
        for t in only_tables_cfg
        if str(t).strip()
    }
    assume(len(only_tables) > 0)

    # 复现 load() 中的过滤逻辑
    filtered_tables = set()
    for dwd_table in DwdLoadTask.TABLE_MAP:
        table_base = dwd_table.split(".")[-1]  # 等价于 _table_base
        if dwd_table.lower() in only_tables or table_base.lower() in only_tables:
            filtered_tables.add(dwd_table)

    # 属性：过滤结果应等于 TABLE_MAP 键与 only_tables 的交集
    # （交集需同时考虑全名和 base name 两种匹配方式）
    expected = set()
    for dwd_table in DwdLoadTask.TABLE_MAP:
        table_base = dwd_table.split(".")[-1]
        if dwd_table.lower() in only_tables or table_base.lower() in only_tables:
            expected.add(dwd_table)

    assert filtered_tables == expected, (
        f"过滤结果不一致: filtered={filtered_tables}, expected={expected}, "
        f"only_tables={only_tables}"
    )

    # 额外属性：过滤结果是 TABLE_MAP 键集合的子集
    assert filtered_tables.issubset(set(DwdLoadTask.TABLE_MAP.keys())), (
        f"过滤结果包含不在 TABLE_MAP 中的表: "
        f"{filtered_tables - set(DwdLoadTask.TABLE_MAP.keys())}"
    )

    # 额外属性：如果 only_tables 中的某个值完全不匹配任何 TABLE_MAP 键，
    # 则该值不应导致任何表被选中
    for cfg_val in only_tables:
        matches_any = any(
            dwd_table.lower() == cfg_val or dwd_table.split(".")[-1].lower() == cfg_val
            for dwd_table in DwdLoadTask.TABLE_MAP
        )
        if not matches_any:
            # 该配置值不应导致任何表被选中
            for t in filtered_tables:
                assert t.lower() != cfg_val and t.split(".")[-1].lower() != cfg_val


# ===========================================================================
# Property 8: DWS 分段累加一致性
# Feature: etl-flow-debug, Property 8: DWS 分段累加一致性
# Validates: Requirements 3.3
#
# 验证策略：直接测试 BaseTask._accumulate_counts 静态方法，
# 对任意分段计数列表，累加结果应等于逐键求和。
# ===========================================================================

# 策略：生成分段计数字典列表
_count_key = st.sampled_from([
    "inserted", "updated", "skipped", "errors", "fetched",
    "processed", "deleted", "merged",
])

_count_value = st.integers(min_value=0, max_value=100000)

_single_counts = st.dictionaries(
    keys=_count_key,
    values=_count_value,
    min_size=1,
    max_size=6,
)

_segment_counts_list = st.lists(
    _single_counts,
    min_size=1,
    max_size=10,
)


@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
@given(segments=_segment_counts_list)
def test_property8_accumulate_counts_consistency(segments):
    """对任意分段计数列表，_accumulate_counts 累加结果应等于逐键求和。

    **Validates: Requirements 3.3**

    属性：对每个数值型键，累加结果 == sum(各分段中该键的值)
    """
    # 使用 _accumulate_counts 逐段累加
    total_via_method: dict = {}
    for seg_counts in segments:
        BaseTask._accumulate_counts(total_via_method, seg_counts)

    # 手动逐键求和作为期望值
    expected: dict = {}
    for seg_counts in segments:
        for key, value in seg_counts.items():
            if isinstance(value, (int, float)):
                expected[key] = expected.get(key, 0) + value
            else:
                expected.setdefault(key, value)

    # 核心属性：两种方式的结果应完全一致
    assert total_via_method == expected, (
        f"累加结果不一致:\n"
        f"  _accumulate_counts: {total_via_method}\n"
        f"  手动逐键求和:       {expected}\n"
        f"  输入分段: {segments}"
    )

    # 额外属性：对每个数值型键，累加值 >= 0（因为输入值 >= 0）
    for key, value in total_via_method.items():
        if isinstance(value, (int, float)):
            assert value >= 0, f"累加值 {key}={value} 不应为负"
