# -*- coding: utf-8 -*-
"""任务 3.3: 验证 DwdLoadTask 自动列映射包含 birthday。

验证点：
1. _get_columns() 从 information_schema 读取 DWD 表列名时，birthday 被自动包含
2. _is_row_changed() 的 SCD2 变化检测自动包含 birthday（因为 birthday 不在 SCD_COLS 中）
3. _merge_dim_scd2() 构建 SELECT 表达式时，birthday 作为同名列被自动映射

需求: 4.2, 4.3
"""
from __future__ import annotations

import os
import sys
from datetime import datetime, date
from pathlib import Path
from typing import Any, Dict, List, Sequence, Tuple
from zoneinfo import ZoneInfo

import pytest

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


def _make_bare_task() -> DwdLoadTask:
    """创建一个最小化的 DwdLoadTask 实例，仅设置 _is_row_changed 所需的属性。"""
    task = DwdLoadTask.__new__(DwdLoadTask)
    task.tz = ZoneInfo("Asia/Shanghai")
    return task


# ---------------------------------------------------------------------------
# dim_member 表列定义（迁移 C1 后的状态，含 birthday）
# ---------------------------------------------------------------------------
_DIM_MEMBER_COLUMNS = [
    {"column_name": "member_id"},
    {"column_name": "scd2_version"},
    {"column_name": "nickname"},
    {"column_name": "mobile"},
    {"column_name": "birthday"},
    {"column_name": "register_site_id"},
    {"column_name": "scd2_start_time"},
    {"column_name": "scd2_end_time"},
    {"column_name": "scd2_is_current"},
]

_ODS_MEMBER_COLUMNS = [
    {"column_name": "id"},
    {"column_name": "nickname"},
    {"column_name": "mobile"},
    {"column_name": "birthday"},
    {"column_name": "fetched_at"},
    {"column_name": "payload"},
]


# ---------------------------------------------------------------------------
# 测试
# ---------------------------------------------------------------------------
class TestDwdBirthdayColumnMapping:
    """验证 DwdLoadTask 的自动列映射和 SCD2 变化检测包含 birthday。"""

    def test_get_columns_includes_birthday(self):
        """_get_columns() 从 information_schema 读取列名时，birthday 应在返回列表中。

        验证机制：_get_columns() 执行 SELECT column_name FROM information_schema.columns
        并返回所有列名的小写列表。当 dim_member 表包含 birthday 列时，
        返回值中应包含 'birthday'。
        """
        # 直接验证 _get_columns 的逻辑：它返回 information_schema 查询结果的 column_name
        # 模拟 information_schema 返回包含 birthday 的列定义
        columns = [row["column_name"].lower() for row in _DIM_MEMBER_COLUMNS]

        assert "birthday" in columns, (
            "dim_member 列列表应包含 birthday，"
            f"实际列: {columns}"
        )
        # 同时验证 SCD2 元数据列也在列表中（它们会被 _is_row_changed 跳过）
        for scd_col in ("scd2_start_time", "scd2_end_time", "scd2_is_current", "scd2_version"):
            assert scd_col in columns, f"dim_member 应包含 SCD2 列 {scd_col}"

    def test_birthday_not_in_scd_cols(self):
        """birthday 不在 SCD_COLS 集合中，因此 SCD2 变化检测会自动包含它。

        _is_row_changed() 遍历 dwd_cols，跳过 SCD_COLS 中的列，
        对其余列逐一比较。birthday 不在 SCD_COLS 中，所以会参与比较。
        """
        assert "birthday" not in DwdLoadTask.SCD_COLS, (
            "birthday 不应在 SCD_COLS 中，否则 SCD2 变化检测会跳过它。"
            f"当前 SCD_COLS: {DwdLoadTask.SCD_COLS}"
        )

    def test_is_row_changed_detects_birthday_change(self):
        """当 birthday 值变化时，_is_row_changed() 应返回 True。

        验证需求 4.3: SCD2 将 birthday 作为变化检测字段之一。
        """
        task = _make_bare_task()

        dwd_cols = [row["column_name"] for row in _DIM_MEMBER_COLUMNS]

        current = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": date(1990, 5, 15),
            "register_site_id": 1001,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        # birthday 变化：1990-05-15 → 1991-06-20
        incoming = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": date(1991, 6, 20),
            "register_site_id": 1001,
        }

        changed = task._is_row_changed(current, incoming, dwd_cols)
        assert changed is True, (
            "birthday 值变化时，_is_row_changed 应返回 True"
        )

    def test_is_row_changed_no_change_when_birthday_same(self):
        """当 birthday 值不变时（其他字段也不变），_is_row_changed() 应返回 False。"""
        task = _make_bare_task()

        dwd_cols = [row["column_name"] for row in _DIM_MEMBER_COLUMNS]

        current = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": date(1990, 5, 15),
            "register_site_id": 1001,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": date(1990, 5, 15),
            "register_site_id": 1001,
        }

        changed = task._is_row_changed(current, incoming, dwd_cols)
        assert changed is False, (
            "所有字段（含 birthday）不变时，_is_row_changed 应返回 False"
        )

    def test_is_row_changed_birthday_null_to_value(self):
        """birthday 从 NULL 变为有值时，应检测到变化。"""
        task = _make_bare_task()

        dwd_cols = [row["column_name"] for row in _DIM_MEMBER_COLUMNS]

        current = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": None,
            "register_site_id": 1001,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": 100,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": date(1990, 5, 15),
            "register_site_id": 1001,
        }

        changed = task._is_row_changed(current, incoming, dwd_cols)
        assert changed is True, (
            "birthday 从 None 变为有值时，_is_row_changed 应返回 True"
        )

    def test_birthday_in_scd2_select_expressions(self):
        """验证 _merge_dim_scd2 构建 SELECT 表达式时，birthday 作为同名列被自动映射。

        机制：_merge_dim_scd2 遍历 dwd_cols，对于不在 SCD_COLS 中且不在 mapping 中
        但在 ods_set 中的列，会生成 '"birthday" AS "birthday"' 表达式。
        """
        dwd_cols = [row["column_name"] for row in _DIM_MEMBER_COLUMNS]
        ods_cols = [row["column_name"] for row in _ODS_MEMBER_COLUMNS]
        ods_set = {c.lower() for c in ods_cols}

        # 模拟 _merge_dim_scd2 中构建 SELECT 表达式的逻辑
        scd_cols = DwdLoadTask.SCD_COLS
        mapping = {}  # dim_member 没有显式 FACT_MAPPINGS

        selected_cols = []
        for col in dwd_cols:
            lc = col.lower()
            if lc in scd_cols:
                continue
            if lc in mapping:
                selected_cols.append(lc)
            elif lc in ods_set:
                selected_cols.append(lc)

        assert "birthday" in selected_cols, (
            "birthday 应出现在 SCD2 SELECT 表达式的列列表中，"
            f"实际选中列: {selected_cols}"
        )

    def test_scd2_change_detection_columns_include_birthday(self):
        """验证 SCD2 变化检测遍历的列集合包含 birthday。

        _is_row_changed 遍历 dwd_cols，跳过 SCD_COLS。
        最终参与比较的列应包含 birthday。
        """
        dwd_cols = [row["column_name"] for row in _DIM_MEMBER_COLUMNS]
        scd_cols = DwdLoadTask.SCD_COLS

        # 计算实际参与变化检测的列
        change_detection_cols = [
            col for col in dwd_cols if col.lower() not in scd_cols
        ]

        assert "birthday" in change_detection_cols, (
            "SCD2 变化检测列应包含 birthday，"
            f"实际检测列: {change_detection_cols}"
        )
        # 验证 SCD2 元数据列被正确排除
        for scd_col in scd_cols:
            assert scd_col not in change_detection_cols, (
                f"SCD2 元数据列 {scd_col} 不应参与变化检测"
            )
