# -*- coding: utf-8 -*-
"""Property 7: birthday ODS→DWD 装载正确性 — hypothesis 属性测试。

# Feature: etl-aggregation-fix, Property 7: birthday ODS→DWD 装载正确性
# **Validates: Requirements 4.2**
#
# 对于任意 ODS member_profiles 中包含 birthday 值的记录，
# DwdLoadTask 装载后 dwd.dim_member 中对应记录的 birthday 应与 ODS 源值一致。
#
# 测试策略：
# - 不依赖真实数据库，通过模拟 _merge_dim_scd2 的列映射构建逻辑来验证
# - 验证 birthday 列在自动列映射中被正确包含
# - 验证 SCD2 行比较逻辑正确传递 birthday 值
"""
from __future__ import annotations

import os
import sys
from datetime import date, datetime
from pathlib import Path
from typing import Any, Dict, Optional
from zoneinfo import ZoneInfo

import pytest
from hypothesis import given, settings, assume
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


# ── 辅助 ──────────────────────────────────────────────────────

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


# dim_member DWD 列（迁移 C1 后，含 birthday）
_DWD_COLS = [
    "member_id", "nickname", "mobile", "birthday",
    "register_site_id", "pay_money_sum", "recharge_money_sum",
    "scd2_start_time", "scd2_end_time", "scd2_is_current", "scd2_version",
]

# ODS member_profiles 列（含 birthday）
_ODS_COLS = [
    "id", "nickname", "mobile", "birthday",
    "pay_money_sum", "recharge_money_sum",
    "register_site_id", "fetched_at", "payload",
]


# ── 生成策略 ──────────────────────────────────────────────────

# 合法生日日期：1920-01-01 ~ 2020-12-31
_birthday_st = st.dates(
    min_value=date(1920, 1, 1),
    max_value=date(2020, 12, 31),
)

# 可选生日：有值或 None
_optional_birthday_st = st.one_of(st.none(), _birthday_st)

# 会员 ID：正整数
_member_id_st = st.integers(min_value=1, max_value=10**12)

# 昵称
_nickname_st = st.text(min_size=1, max_size=20)

# 手机号
_mobile_st = st.from_regex(r"1[3-9]\d{9}", fullmatch=True)


# ── Property 7: birthday ODS→DWD 装载正确性 ──────────────────
# Feature: etl-aggregation-fix, Property 7: birthday ODS→DWD 装载正确性
# **Validates: Requirements 4.2**


class TestProperty7BirthdayOdsToDwdLoad:
    """Property 7: birthday ODS→DWD 装载正确性。

    对于任意 ODS member_profiles 中包含 birthday 值的记录，
    DwdLoadTask 装载后 dwd.dim_member 中对应记录的 birthday
    应与 ODS 源值一致。
    """

    @given(birthday=_birthday_st)
    @settings(max_examples=150)
    def test_birthday_included_in_column_mapping(self, birthday: date):
        """birthday 列应出现在 _merge_dim_scd2 的自动列映射 SELECT 表达式中。

        机制：_merge_dim_scd2 遍历 dwd_cols，跳过 SCD_COLS，
        对于不在显式 mapping 中但在 ods_set 中的列，生成同名映射。
        birthday 满足此条件，因此应被自动包含。
        """
        ods_set = {c.lower() for c in _ODS_COLS}
        scd_cols = DwdLoadTask.SCD_COLS

        # 获取 dim_member 的显式映射（FACT_MAPPINGS）
        mapping_entries = DwdLoadTask.FACT_MAPPINGS.get("dwd.dim_member", [])
        explicit_mapping = {dst.lower() for dst, _, _ in mapping_entries}

        # 模拟 _merge_dim_scd2 的列选择逻辑
        selected_cols = []
        for col in _DWD_COLS:
            lc = col.lower()
            if lc in scd_cols:
                continue
            if lc in explicit_mapping:
                selected_cols.append(lc)
            elif lc in ods_set:
                selected_cols.append(lc)

        assert "birthday" in selected_cols, (
            f"birthday 应在自动列映射中被选中，实际选中列: {selected_cols}"
        )

    @given(birthday=_birthday_st, member_id=_member_id_st)
    @settings(max_examples=150)
    def test_birthday_value_preserved_in_mapped_row(
        self, birthday: date, member_id: int
    ):
        """ODS 源行中的 birthday 值经过行映射后应保持不变。

        _merge_dim_scd2 从 ODS 读取行后，对每列执行：
        1. 先从 row.get(lc) 取值
        2. 若为 None 且在 mapping 中，再从 mapping 源列取值
        birthday 不在显式 mapping 中，直接从 row.get("birthday") 取值。
        """
        # 模拟 ODS 查询返回的行（已经过 SELECT 表达式处理）
        ods_row = {
            "member_id": member_id,
            "nickname": "测试会员",
            "mobile": "13800138000",
            "birthday": birthday,
            "register_site_id": 1001,
            "pay_money_sum": 0,
            "recharge_money_sum": 0,
        }

        mapping_entries = DwdLoadTask.FACT_MAPPINGS.get("dwd.dim_member", [])
        mapping = {dst.lower(): (src, ct) for dst, src, ct in mapping_entries}
        scd_cols = DwdLoadTask.SCD_COLS

        # 模拟 _merge_dim_scd2 的行映射逻辑
        mapped_row: Dict[str, Any] = {}
        for col in _DWD_COLS:
            lc = col.lower()
            if lc in scd_cols:
                continue
            value = ods_row.get(lc)
            if value is None and lc in mapping:
                src, _ = mapping[lc]
                value = ods_row.get(src.lower())
            mapped_row[lc] = value

        assert mapped_row["birthday"] == birthday, (
            f"映射后 birthday 应为 {birthday}，实际为 {mapped_row['birthday']}"
        )

    @given(
        old_birthday=_optional_birthday_st,
        new_birthday=_birthday_st,
        member_id=_member_id_st,
    )
    @settings(max_examples=150)
    def test_birthday_change_detected_by_scd2(
        self, old_birthday: Optional[date], new_birthday: date, member_id: int
    ):
        """当 ODS birthday 与 DWD 现有值不同时，SCD2 变化检测应识别变更。

        _is_row_changed 遍历 dwd_cols，跳过 SCD_COLS，
        对每列调用 _values_equal 比较。birthday 不在 SCD_COLS 中，
        因此值变化时应返回 True。
        """
        assume(old_birthday != new_birthday)

        task = _make_bare_task()

        current = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": old_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": new_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        changed = task._is_row_changed(current, incoming, _DWD_COLS)
        assert changed is True, (
            f"birthday 从 {old_birthday} 变为 {new_birthday} 时，"
            "_is_row_changed 应返回 True"
        )

    @given(birthday=_birthday_st, member_id=_member_id_st)
    @settings(max_examples=150)
    def test_same_birthday_no_false_change(
        self, birthday: date, member_id: int
    ):
        """当 ODS birthday 与 DWD 现有值相同时，不应触发虚假变更。

        确保 _values_equal 对 date 类型的比较是正确的，
        不会因类型转换导致误判。
        """
        task = _make_bare_task()

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

        changed = task._is_row_changed(current, incoming, _DWD_COLS)
        assert changed is False, (
            f"birthday 值相同 ({birthday}) 时，_is_row_changed 应返回 False"
        )


# ── Property 8: birthday SCD2 变化检测 ─────────────────────────
# Feature: etl-aggregation-fix, Property 8: birthday SCD2 变化检测
# **Validates: Requirements 4.3**
#
# 对于任意 dim_member 现有记录，当 ODS 中同一会员的 birthday 值发生变化时，
# SCD2 应关闭旧版本并创建新版本，新版本的 birthday 等于新值。
#
# 测试策略：
# - 不依赖真实数据库，模拟 _merge_dim_scd2 中的 SCD2 决策逻辑
# - 验证 birthday 变化触发变更检测 → 旧版本关闭 → 新版本创建
# - 验证版本号递增、scd2_is_current 标记、scd2_end_time 填充


class TestProperty8BirthdayScd2ChangeDetection:
    """Property 8: birthday SCD2 变化检测。

    对于任意 dim_member 现有记录，当 ODS 中同一会员的 birthday 值发生变化时，
    SCD2 应关闭旧版本并创建新版本，新版本的 birthday 等于新值。
    """

    @given(
        old_birthday=_optional_birthday_st,
        new_birthday=_birthday_st,
        member_id=_member_id_st,
        old_version=st.integers(min_value=1, max_value=100),
    )
    @settings(max_examples=150)
    def test_birthday_change_triggers_scd2_version_increment(
        self,
        old_birthday: Optional[date],
        new_birthday: date,
        member_id: int,
        old_version: int,
    ):
        """birthday 变化时，新版本号应为旧版本号 + 1。

        模拟 _merge_dim_scd2 中的核心决策：
        1. _is_row_changed 检测到变化
        2. version = current.scd2_version + 1
        3. to_insert 中记录 (incoming, version)
        """
        assume(old_birthday != new_birthday)

        task = _make_bare_task()

        current = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": old_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": old_version,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": new_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        # 步骤 1：变更检测
        changed = task._is_row_changed(current, incoming, _DWD_COLS)
        assert changed is True

        # 步骤 2：模拟 _merge_dim_scd2 的版本号计算逻辑
        new_version = (current.get("scd2_version") or 1) + 1
        assert new_version == old_version + 1, (
            f"新版本号应为 {old_version + 1}，实际为 {new_version}"
        )

    @given(
        old_birthday=_optional_birthday_st,
        new_birthday=_birthday_st,
        member_id=_member_id_st,
        old_version=st.integers(min_value=1, max_value=100),
    )
    @settings(max_examples=150)
    def test_scd2_new_version_carries_updated_birthday(
        self,
        old_birthday: Optional[date],
        new_birthday: date,
        member_id: int,
        old_version: int,
    ):
        """新版本行应携带更新后的 birthday 值。

        模拟 _insert_dim_rows_bulk 的 build_row 逻辑：
        对非 SCD 列从 incoming 行取值，SCD 列由框架填充。
        """
        assume(old_birthday != new_birthday)

        task = _make_bare_task()
        now = datetime(2025, 6, 15, 10, 0, 0)

        current = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": old_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": old_version,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": new_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        # 确认变更被检测到
        assert task._is_row_changed(current, incoming, _DWD_COLS) is True

        # 模拟 _insert_dim_rows_bulk 的 build_row 逻辑
        new_version = (current.get("scd2_version") or 1) + 1
        sorted_cols = sorted(c.lower() for c in _DWD_COLS)
        new_row_values: Dict[str, Any] = {}
        for c in sorted_cols:
            if c == "scd2_start_time":
                new_row_values[c] = now
            elif c == "scd2_end_time":
                new_row_values[c] = datetime(9999, 12, 31, 0, 0, 0)
            elif c == "scd2_is_current":
                new_row_values[c] = 1
            elif c == "scd2_version":
                new_row_values[c] = new_version
            else:
                new_row_values[c] = incoming.get(c)

        # 核心断言：新版本的 birthday 等于新值
        assert new_row_values["birthday"] == new_birthday, (
            f"新版本 birthday 应为 {new_birthday}，实际为 {new_row_values['birthday']}"
        )
        # 新版本标记为当前版本
        assert new_row_values["scd2_is_current"] == 1
        assert new_row_values["scd2_version"] == old_version + 1

    @given(
        old_birthday=_optional_birthday_st,
        new_birthday=_birthday_st,
        member_id=_member_id_st,
    )
    @settings(max_examples=150)
    def test_scd2_old_version_closure_pattern(
        self,
        old_birthday: Optional[date],
        new_birthday: date,
        member_id: int,
    ):
        """birthday 变化时，旧版本应被关闭（scd2_is_current=0, scd2_end_time=now）。

        模拟 _merge_dim_scd2 的决策流程：
        - _is_row_changed 返回 True → pk_key 加入 to_close
        - _close_current_dim_bulk 将旧版本 scd2_is_current 设为 0
        """
        assume(old_birthday != new_birthday)

        task = _make_bare_task()
        now = datetime(2025, 6, 15, 10, 0, 0)

        current = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": old_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }
        incoming = {
            "member_id": member_id,
            "nickname": "张三",
            "mobile": "13800138000",
            "birthday": new_birthday,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        # 模拟 _merge_dim_scd2 的完整决策流程
        business_keys = ["member_id"]
        pk_key = tuple(incoming.get(pk) for pk in business_keys)
        current_by_pk = {pk_key: current}

        to_close: list = []
        to_insert: list = []

        changed = task._is_row_changed(current, incoming, _DWD_COLS)
        if changed:
            version = (current.get("scd2_version") or 1) + 1
            to_close.append(pk_key)
            to_insert.append((incoming, version))

        # 旧版本应在关闭列表中
        assert len(to_close) == 1, "birthday 变化应触发旧版本关闭"
        assert to_close[0] == (member_id,)

        # 新版本应在插入列表中
        assert len(to_insert) == 1, "birthday 变化应触发新版本插入"
        inserted_row, inserted_version = to_insert[0]
        assert inserted_row["birthday"] == new_birthday
        assert inserted_version == 2

        # 模拟关闭后的旧版本状态
        closed_current = {**current, "scd2_is_current": 0, "scd2_end_time": now}
        assert closed_current["scd2_is_current"] == 0
        assert closed_current["scd2_end_time"] == now
        assert closed_current["birthday"] == old_birthday  # 旧版本保留原 birthday


# ── Property 9: 生日 UPSERT 幂等性 ─────────────────────────────
# Feature: etl-aggregation-fix, Property 9: 生日 UPSERT 幂等性
# **Validates: Requirements 5.2, 5.5**
#
# 对于任意 (member_id, assistant_id) 组合，连续两次提交不同的 birthday_value，
# member_birthday_manual 表中该组合应只有一条记录，
# 且 birthday_value 等于最后一次提交的值。
#
# 测试策略：
# - 纯单元测试，不依赖真实数据库
# - 用内存字典模拟 member_birthday_manual 表的 UPSERT 语义
#   （INSERT ON CONFLICT (member_id, recorded_by_assistant_id) DO UPDATE）
# - 验证同一 (member_id, assistant_id) 连续两次提交后只保留一条记录
# - 验证不同 assistant_id 对同一 member_id 提交时各自保留独立记录


def _upsert_birthday(
    table: dict,
    member_id: int,
    birthday_value: date,
    assistant_id: int,
    assistant_name: str,
) -> None:
    """模拟后端 UPSERT SQL 语义：

    INSERT INTO member_birthday_manual
        (member_id, birthday_value, recorded_by_assistant_id, recorded_by_name)
    VALUES (%s, %s, %s, %s)
    ON CONFLICT (member_id, recorded_by_assistant_id)
    DO UPDATE SET
        birthday_value = EXCLUDED.birthday_value,
        recorded_at    = NOW()

    table 以 (member_id, assistant_id) 为键，值为完整行字典。
    """
    key = (member_id, assistant_id)
    if key in table:
        # ON CONFLICT → DO UPDATE
        table[key]["birthday_value"] = birthday_value
        table[key]["recorded_at"] = datetime.now(tz=ZoneInfo("Asia/Shanghai"))
    else:
        # INSERT
        table[key] = {
            "member_id": member_id,
            "birthday_value": birthday_value,
            "recorded_by_assistant_id": assistant_id,
            "recorded_by_name": assistant_name,
            "recorded_at": datetime.now(tz=ZoneInfo("Asia/Shanghai")),
            "source": "assistant",
        }


# ── hypothesis 策略 ──────────────────────────────────────────

# 助教 ID：正整数
_assistant_id_st = st.integers(min_value=1, max_value=10**6)

# 助教姓名：中文或英文，1~50 字符
_assistant_name_st = st.text(
    alphabet=st.sampled_from(
        "张王李赵刘陈杨黄周吴徐孙马胡朱郭何罗高林"
        "abcdefghijklmnopqrstuvwxyz"
    ),
    min_size=1,
    max_size=50,
)


class TestProperty9BirthdayUpsertIdempotency:
    """Property 9: 生日 UPSERT 幂等性。

    对于任意 (member_id, assistant_id) 组合，连续两次提交不同的 birthday_value，
    member_birthday_manual 表中该组合应只有一条记录，
    且 birthday_value 等于最后一次提交的值。
    """

    @given(
        member_id=_member_id_st,
        assistant_id=_assistant_id_st,
        birthday_first=_birthday_st,
        birthday_second=_birthday_st,
        assistant_name=_assistant_name_st,
    )
    @settings(max_examples=150)
    def test_upsert_same_key_keeps_single_record(
        self,
        member_id: int,
        assistant_id: int,
        birthday_first: date,
        birthday_second: date,
        assistant_name: str,
    ):
        """同一 (member_id, assistant_id) 连续两次提交后，表中只有一条记录，
        且 birthday_value 等于最后一次提交的值。
        """
        assume(birthday_first != birthday_second)

        table: dict = {}

        # 第一次提交
        _upsert_birthday(table, member_id, birthday_first, assistant_id, assistant_name)
        assert len(table) == 1
        assert table[(member_id, assistant_id)]["birthday_value"] == birthday_first

        # 第二次提交（不同 birthday）
        _upsert_birthday(table, member_id, birthday_second, assistant_id, assistant_name)

        # 核心断言：仍然只有一条记录
        assert len(table) == 1, (
            f"同一 (member_id={member_id}, assistant_id={assistant_id}) "
            f"两次提交后应只有 1 条记录，实际 {len(table)} 条"
        )
        # 核心断言：值为最后一次提交
        assert table[(member_id, assistant_id)]["birthday_value"] == birthday_second, (
            f"birthday_value 应为最后提交的 {birthday_second}，"
            f"实际为 {table[(member_id, assistant_id)]['birthday_value']}"
        )

    @given(
        member_id=_member_id_st,
        assistant_id=_assistant_id_st,
        birthday_value=_birthday_st,
        assistant_name=_assistant_name_st,
    )
    @settings(max_examples=150)
    def test_upsert_same_value_is_idempotent(
        self,
        member_id: int,
        assistant_id: int,
        birthday_value: date,
        assistant_name: str,
    ):
        """同一 (member_id, assistant_id) 提交相同 birthday_value 两次，
        结果仍为一条记录且值不变（幂等性）。
        """
        table: dict = {}

        _upsert_birthday(table, member_id, birthday_value, assistant_id, assistant_name)
        _upsert_birthday(table, member_id, birthday_value, assistant_id, assistant_name)

        assert len(table) == 1
        assert table[(member_id, assistant_id)]["birthday_value"] == birthday_value

    @given(
        member_id=_member_id_st,
        assistant_id_a=_assistant_id_st,
        assistant_id_b=_assistant_id_st,
        birthday_a=_birthday_st,
        birthday_b=_birthday_st,
        name_a=_assistant_name_st,
        name_b=_assistant_name_st,
    )
    @settings(max_examples=150)
    def test_different_assistants_keep_independent_records(
        self,
        member_id: int,
        assistant_id_a: int,
        assistant_id_b: int,
        birthday_a: date,
        birthday_b: date,
        name_a: str,
        name_b: str,
    ):
        """不同 assistant_id 对同一 member_id 提交时，各自保留独立记录。

        唯一约束为 (member_id, recorded_by_assistant_id)，
        不同助教的提交互不影响。
        """
        assume(assistant_id_a != assistant_id_b)

        table: dict = {}

        _upsert_birthday(table, member_id, birthday_a, assistant_id_a, name_a)
        _upsert_birthday(table, member_id, birthday_b, assistant_id_b, name_b)

        # 两个助教各一条记录
        assert len(table) == 2, (
            f"不同助教提交后应有 2 条记录，实际 {len(table)} 条"
        )
        assert table[(member_id, assistant_id_a)]["birthday_value"] == birthday_a
        assert table[(member_id, assistant_id_b)]["birthday_value"] == birthday_b

        # 再次用 assistant_id_a 提交新值，不影响 assistant_id_b
        new_birthday_a = birthday_b  # 用不同的值
        _upsert_birthday(table, member_id, new_birthday_a, assistant_id_a, name_a)

        assert len(table) == 2, "更新 assistant_a 不应影响 assistant_b 的记录数"
        assert table[(member_id, assistant_id_a)]["birthday_value"] == new_birthday_a
        assert table[(member_id, assistant_id_b)]["birthday_value"] == birthday_b


# ── Property 10: 手动补录优先于 API 来源 ──────────────────────────
# Feature: etl-aggregation-fix, Property 10: 手动补录优先于 API 来源
# **Validates: Requirements 5.4**
#
# 对于任意同时在 dim_member.birthday 和 member_birthday_manual 中有值的会员，
# DWS 任务输出的 member_birthday 应等于手动补录表中的值。
#
# 测试策略：
# - 纯单元测试，不依赖真实数据库
# - 用 Python 函数模拟 DWS 任务 _extract_member_info() 中的 COALESCE SQL 语义：
#     COALESCE(
#         (SELECT birthday_value FROM fdw_app.member_birthday_manual
#          WHERE member_id = m.member_id ORDER BY recorded_at ASC LIMIT 1),
#         m.birthday
#     ) AS member_birthday
# - 验证四种场景：
#   a. 两者都有值 → 输出手动补录值
#   b. 只有 API 值 → 输出 API 值
#   c. 只有手动补录值 → 输出手动补录值
#   d. 两者都为 None → 输出 None


def _coalesce_birthday(
    api_birthday: Optional[date],
    manual_records: list[dict],
) -> Optional[date]:
    """模拟 DWS 任务 _extract_member_info() 中的 COALESCE 生日读取逻辑。

    SQL 语义：
        COALESCE(
            (SELECT birthday_value
             FROM fdw_app.member_birthday_manual
             WHERE member_id = m.member_id
             ORDER BY recorded_at ASC
             LIMIT 1),
            m.birthday
        )

    参数:
        api_birthday: dim_member.birthday（API 来源），可为 None
        manual_records: member_birthday_manual 中该会员的所有记录列表，
                        每条记录为 {"birthday_value": date, "recorded_at": datetime}
                        可为空列表（无手动补录）

    返回:
        优先手动补录值（按 recorded_at ASC 取第一条），其次 API 值
    """
    # 子查询：按 recorded_at ASC 排序取第一条
    if manual_records:
        sorted_records = sorted(manual_records, key=lambda r: r["recorded_at"])
        manual_value = sorted_records[0]["birthday_value"]
    else:
        manual_value = None

    # COALESCE：优先手动补录值，其次 API 值
    return manual_value if manual_value is not None else api_birthday


# ── hypothesis 策略（复用已有 + 新增） ────────────────────────────

# 时间戳策略：用于 recorded_at
_recorded_at_st = st.datetimes(
    min_value=datetime(2024, 1, 1),
    max_value=datetime(2026, 12, 31),
)


class TestProperty10ManualBirthdayPriority:
    """Property 10: 手动补录优先于 API 来源。

    对于任意同时在 dim_member.birthday 和 member_birthday_manual 中有值的会员，
    DWS 任务输出的 member_birthday 应等于手动补录表中的值。
    """

    @given(
        api_birthday=_birthday_st,
        manual_birthday=_birthday_st,
        recorded_at=_recorded_at_st,
    )
    @settings(max_examples=150)
    def test_both_present_manual_wins(
        self,
        api_birthday: date,
        manual_birthday: date,
        recorded_at: datetime,
    ):
        """场景 a：两者都有值时，输出手动补录值。"""
        manual_records = [
            {"birthday_value": manual_birthday, "recorded_at": recorded_at},
        ]

        result = _coalesce_birthday(api_birthday, manual_records)

        assert result == manual_birthday, (
            f"两者都有值时应输出手动补录值 {manual_birthday}，"
            f"实际输出 {result}（API 值为 {api_birthday}）"
        )

    @given(api_birthday=_birthday_st)
    @settings(max_examples=150)
    def test_only_api_returns_api(self, api_birthday: date):
        """场景 b：只有 API 值时，输出 API 值。"""
        manual_records: list[dict] = []

        result = _coalesce_birthday(api_birthday, manual_records)

        assert result == api_birthday, (
            f"只有 API 值时应输出 {api_birthday}，实际输出 {result}"
        )

    @given(
        manual_birthday=_birthday_st,
        recorded_at=_recorded_at_st,
    )
    @settings(max_examples=150)
    def test_only_manual_returns_manual(
        self,
        manual_birthday: date,
        recorded_at: datetime,
    ):
        """场景 c：只有手动补录值时，输出手动补录值。"""
        manual_records = [
            {"birthday_value": manual_birthday, "recorded_at": recorded_at},
        ]

        result = _coalesce_birthday(None, manual_records)

        assert result == manual_birthday, (
            f"只有手动补录值时应输出 {manual_birthday}，实际输出 {result}"
        )

    @settings(max_examples=150)
    @given(data=st.data())
    def test_both_none_returns_none(self, data: st.DataObject):
        """场景 d：两者都为 None 时，输出 None。"""
        manual_records: list[dict] = []

        result = _coalesce_birthday(None, manual_records)

        assert result is None, (
            f"两者都为 None 时应输出 None，实际输出 {result}"
        )

    @given(
        api_birthday=_birthday_st,
        manual_birthdays=st.lists(
            _birthday_st, min_size=2, max_size=5,
        ),
        recorded_ats=st.lists(
            _recorded_at_st, min_size=2, max_size=5,
        ),
    )
    @settings(max_examples=150)
    def test_multiple_manual_records_earliest_wins(
        self,
        api_birthday: date,
        manual_birthdays: list[date],
        recorded_ats: list[datetime],
    ):
        """多条手动补录记录时，按 recorded_at ASC 取第一条（最早提交优先）。

        SQL: ORDER BY recorded_at ASC LIMIT 1
        """
        # 确保列表长度一致
        size = min(len(manual_birthdays), len(recorded_ats))
        assume(size >= 2)
        manual_birthdays = manual_birthdays[:size]
        recorded_ats = recorded_ats[:size]

        # 确保 recorded_at 不完全相同（否则排序不确定）
        assume(len(set(recorded_ats)) == len(recorded_ats))

        manual_records = [
            {"birthday_value": bday, "recorded_at": rat}
            for bday, rat in zip(manual_birthdays, recorded_ats)
        ]

        result = _coalesce_birthday(api_birthday, manual_records)

        # 期望值：按 recorded_at ASC 排序后第一条的 birthday_value
        expected = sorted(manual_records, key=lambda r: r["recorded_at"])[0]["birthday_value"]

        assert result == expected, (
            f"多条手动补录时应取最早提交的 {expected}，实际输出 {result}"
        )


# ── Property 11: SCD2 更新不影响手动补录表 ──────────────────────────
# Feature: etl-aggregation-fix, Property 11: SCD2 更新不影响手动补录表
# **Validates: Requirements 5.6**
#
# 对于任意在 member_birthday_manual 中有记录的会员，
# 执行 DwdLoadTask SCD2 更新 dim_member.birthday 后，
# member_birthday_manual 中的记录应保持不变。
#
# 测试策略：
# - 纯单元测试，不依赖真实数据库
# - 核心思路：SCD2 更新操作只涉及 dim_member 表（ETL 库），
#   不会触及 member_birthday_manual 表（业务库）
# - 模拟场景：
#   a. 手动补录表中有记录（用字典模拟）
#   b. 执行 SCD2 变更检测（使用 DwdLoadTask._is_row_changed）
#   c. 模拟 SCD2 关闭旧版本 + 插入新版本
#   d. 验证手动补录表的记录完全不变
# - 验证 SCD2 操作的 to_close 和 to_insert 列表中不包含任何对
#   member_birthday_manual 的操作


class TestProperty11Scd2DoesNotAffectManualTable:
    """Property 11: SCD2 更新不影响手动补录表。

    对于任意在 member_birthday_manual 中有记录的会员，
    执行 DwdLoadTask SCD2 更新 dim_member.birthday 后，
    member_birthday_manual 中的记录应保持不变。
    """

    @given(
        member_id=_member_id_st,
        api_birthday_old=_optional_birthday_st,
        api_birthday_new=_birthday_st,
        manual_birthday=_birthday_st,
        assistant_id=_assistant_id_st,
        assistant_name=_assistant_name_st,
        recorded_at=_recorded_at_st,
    )
    @settings(max_examples=150)
    def test_scd2_update_preserves_manual_records(
        self,
        member_id: int,
        api_birthday_old: Optional[date],
        api_birthday_new: date,
        manual_birthday: date,
        assistant_id: int,
        assistant_name: str,
        recorded_at: datetime,
    ):
        """SCD2 更新 dim_member.birthday 后，手动补录表记录完全不变。

        流程：
        1. 手动补录表中已有该会员的记录
        2. API 返回新的 birthday → 触发 SCD2 变更
        3. SCD2 关闭旧版本 + 插入新版本（仅操作 dim_member）
        4. 手动补录表的记录应与操作前完全一致
        """
        assume(api_birthday_old != api_birthday_new)

        task = _make_bare_task()

        # ── 1. 手动补录表初始状态（深拷贝用于后续比较） ──
        manual_table: dict = {}
        _upsert_birthday(manual_table, member_id, manual_birthday, assistant_id, assistant_name)
        # 覆盖 recorded_at 为确定性值
        manual_table[(member_id, assistant_id)]["recorded_at"] = recorded_at

        # 保存快照用于最终比较
        snapshot_key = (member_id, assistant_id)
        snapshot = {
            "member_id": manual_table[snapshot_key]["member_id"],
            "birthday_value": manual_table[snapshot_key]["birthday_value"],
            "recorded_by_assistant_id": manual_table[snapshot_key]["recorded_by_assistant_id"],
            "recorded_by_name": manual_table[snapshot_key]["recorded_by_name"],
            "recorded_at": manual_table[snapshot_key]["recorded_at"],
            "source": manual_table[snapshot_key]["source"],
        }

        # ── 2. dim_member 当前行（SCD2 旧版本） ──
        current_dim = {
            "member_id": member_id,
            "nickname": "测试会员",
            "mobile": "13800138000",
            "birthday": api_birthday_old,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }

        # ── 3. ODS 传入行（API 新值） ──
        incoming = {
            "member_id": member_id,
            "nickname": "测试会员",
            "mobile": "13800138000",
            "birthday": api_birthday_new,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        # ── 4. 执行 SCD2 变更检测 ──
        changed = task._is_row_changed(current_dim, incoming, _DWD_COLS)
        assert changed is True, "birthday 变化应被 SCD2 检测到"

        # ── 5. 模拟 SCD2 关闭旧版本 + 插入新版本 ──
        to_close = []
        to_insert = []

        pk_key = (member_id,)
        version = (current_dim.get("scd2_version") or 1) + 1
        to_close.append(pk_key)
        to_insert.append((incoming, version))

        # ── 6. 核心断言：手动补录表完全不变 ──
        assert len(manual_table) == 1, (
            f"SCD2 操作后手动补录表应仍有 1 条记录，实际 {len(manual_table)} 条"
        )
        current_record = manual_table[snapshot_key]
        assert current_record["member_id"] == snapshot["member_id"], (
            "SCD2 操作不应改变手动补录表的 member_id"
        )
        assert current_record["birthday_value"] == snapshot["birthday_value"], (
            f"SCD2 操作不应改变手动补录表的 birthday_value，"
            f"期望 {snapshot['birthday_value']}，实际 {current_record['birthday_value']}"
        )
        assert current_record["recorded_by_assistant_id"] == snapshot["recorded_by_assistant_id"], (
            "SCD2 操作不应改变手动补录表的 recorded_by_assistant_id"
        )
        assert current_record["recorded_by_name"] == snapshot["recorded_by_name"], (
            "SCD2 操作不应改变手动补录表的 recorded_by_name"
        )
        assert current_record["recorded_at"] == snapshot["recorded_at"], (
            "SCD2 操作不应改变手动补录表的 recorded_at"
        )
        assert current_record["source"] == snapshot["source"], (
            "SCD2 操作不应改变手动补录表的 source"
        )

    @given(
        member_id=_member_id_st,
        api_birthday_old=_optional_birthday_st,
        api_birthday_new=_birthday_st,
        manual_birthday=_birthday_st,
        assistant_id=_assistant_id_st,
    )
    @settings(max_examples=150)
    def test_scd2_to_close_and_to_insert_only_target_dim_member(
        self,
        member_id: int,
        api_birthday_old: Optional[date],
        api_birthday_new: date,
        manual_birthday: date,
        assistant_id: int,
    ):
        """SCD2 的 to_close 和 to_insert 操作列表仅涉及 dim_member 的主键，
        不包含任何对 member_birthday_manual 的操作。

        验证 SCD2 操作的"作用域"严格限定在 dim_member 表。
        """
        assume(api_birthday_old != api_birthday_new)

        task = _make_bare_task()

        # dim_member 当前行
        current_dim = {
            "member_id": member_id,
            "nickname": "测试会员",
            "mobile": "13800138000",
            "birthday": api_birthday_old,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
            "scd2_version": 1,
            "scd2_start_time": datetime(2025, 1, 1),
            "scd2_end_time": datetime(9999, 12, 31),
            "scd2_is_current": 1,
        }

        incoming = {
            "member_id": member_id,
            "nickname": "测试会员",
            "mobile": "13800138000",
            "birthday": api_birthday_new,
            "register_site_id": 1001,
            "pay_money_sum": 100,
            "recharge_money_sum": 50,
        }

        # 执行 SCD2 变更检测
        changed = task._is_row_changed(current_dim, incoming, _DWD_COLS)
        assert changed is True

        # 模拟 SCD2 决策
        to_close = []
        to_insert = []
        pk_key = (member_id,)
        version = (current_dim.get("scd2_version") or 1) + 1
        to_close.append(pk_key)
        to_insert.append((incoming, version))

        # 核心断言：to_close 中的主键仅为 dim_member 的 member_id
        for pk in to_close:
            assert len(pk) == 1, "SCD2 关闭操作的主键应为单元素元组 (member_id,)"
            assert pk == (member_id,), "SCD2 关闭操作应仅针对 dim_member 的 member_id"

        # 核心断言：to_insert 中的行仅包含 dim_member 的列
        dim_member_cols = set(c.lower() for c in _DWD_COLS if c.lower() not in DwdLoadTask.SCD_COLS)
        manual_table_cols = {
            "recorded_by_assistant_id", "recorded_by_name",
            "recorded_at", "source", "birthday_value",
        }
        for row, ver in to_insert:
            row_keys = set(row.keys())
            # 插入行不应包含 member_birthday_manual 的专属列
            overlap = row_keys & manual_table_cols
            assert not overlap, (
                f"SCD2 插入行不应包含 member_birthday_manual 的列，"
                f"但发现: {overlap}"
            )

    @given(
        member_ids=st.lists(
            _member_id_st, min_size=2, max_size=5, unique=True,
        ),
        api_birthdays_old=st.lists(
            _optional_birthday_st, min_size=2, max_size=5,
        ),
        api_birthdays_new=st.lists(
            _birthday_st, min_size=2, max_size=5,
        ),
        manual_birthdays=st.lists(
            _birthday_st, min_size=2, max_size=5,
        ),
        assistant_ids=st.lists(
            _assistant_id_st, min_size=2, max_size=5,
        ),
        assistant_names=st.lists(
            _assistant_name_st, min_size=2, max_size=5,
        ),
    )
    @settings(max_examples=150)
    def test_batch_scd2_updates_preserve_all_manual_records(
        self,
        member_ids: list[int],
        api_birthdays_old: list[Optional[date]],
        api_birthdays_new: list[date],
        manual_birthdays: list[date],
        assistant_ids: list[int],
        assistant_names: list[str],
    ):
        """批量 SCD2 更新多个会员时，所有手动补录记录均保持不变。

        模拟真实场景：一次 DwdLoadTask 执行可能更新多个会员的 dim_member，
        手动补录表中可能有多条记录，全部应不受影响。
        """
        size = min(
            len(member_ids), len(api_birthdays_old), len(api_birthdays_new),
            len(manual_birthdays), len(assistant_ids), len(assistant_names),
        )
        assume(size >= 2)

        # 确保至少有一个会员的 birthday 发生变化
        has_change = False
        for i in range(size):
            if api_birthdays_old[i] != api_birthdays_new[i]:
                has_change = True
                break
        assume(has_change)

        task = _make_bare_task()

        # ── 构建手动补录表 ──
        manual_table: dict = {}
        for i in range(size):
            _upsert_birthday(
                manual_table,
                member_ids[i],
                manual_birthdays[i],
                assistant_ids[i],
                assistant_names[i],
            )

        # 保存快照
        manual_snapshot = {
            k: {**v} for k, v in manual_table.items()
        }

        # ── 批量执行 SCD2 更新 ──
        to_close = []
        to_insert = []
        for i in range(size):
            current_dim = {
                "member_id": member_ids[i],
                "nickname": "会员" + str(i),
                "mobile": "13800138" + str(i).zfill(3),
                "birthday": api_birthdays_old[i],
                "register_site_id": 1001,
                "pay_money_sum": 0,
                "recharge_money_sum": 0,
                "scd2_version": 1,
                "scd2_start_time": datetime(2025, 1, 1),
                "scd2_end_time": datetime(9999, 12, 31),
                "scd2_is_current": 1,
            }
            incoming = {
                "member_id": member_ids[i],
                "nickname": "会员" + str(i),
                "mobile": "13800138" + str(i).zfill(3),
                "birthday": api_birthdays_new[i],
                "register_site_id": 1001,
                "pay_money_sum": 0,
                "recharge_money_sum": 0,
            }

            changed = task._is_row_changed(current_dim, incoming, _DWD_COLS)
            if changed:
                pk_key = (member_ids[i],)
                version = (current_dim.get("scd2_version") or 1) + 1
                to_close.append(pk_key)
                to_insert.append((incoming, version))

        # ── 核心断言：手动补录表所有记录完全不变 ──
        assert len(manual_table) == len(manual_snapshot), (
            f"SCD2 批量更新后手动补录表记录数应不变，"
            f"期望 {len(manual_snapshot)}，实际 {len(manual_table)}"
        )
        for key in manual_snapshot:
            assert key in manual_table, (
                f"手动补录表中键 {key} 在 SCD2 操作后不应消失"
            )
            for field in ("member_id", "birthday_value", "recorded_by_assistant_id",
                          "recorded_by_name", "source"):
                assert manual_table[key][field] == manual_snapshot[key][field], (
                    f"手动补录表键 {key} 的 {field} 在 SCD2 操作后不应改变，"
                    f"期望 {manual_snapshot[key][field]}，实际 {manual_table[key][field]}"
                )
