# -*- coding: utf-8 -*-
"""跨店会员可查属性测试 — 使用 hypothesis 验证多门店场景下会员维度信息的可达性。

纯单元测试，使用 FakeDBOperations 模拟数据库查询，不涉及真实数据库连接。

# Feature: etl-aggregation-fix, Property 6: 跨店会员可查
# **Validates: Requirements 3.1, 3.2**
"""
from __future__ import annotations

import re
from typing import Any, Dict, List, Set, Tuple

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


# ── 核心逻辑：模拟"通过事实表反查 dim_member"的查询行为 ──────────
#
# 实际 SQL 模式：
#   SELECT ... FROM dwd.dim_member
#   WHERE member_id IN (
#       SELECT DISTINCT tenant_member_id
#       FROM dwd.{fact_table}
#       WHERE site_id = %s AND tenant_member_id IS NOT NULL AND tenant_member_id != 0
#   ) AND scd2_is_current = 1
#
# 属性要验证的是：无论会员在哪个门店注册，只要在目标门店有消费记录（事实表中存在），
# 该查询模式就能返回其维度信息。


def simulate_fact_table_lookup(
    fact_rows: List[Dict[str, Any]],
    dim_members: List[Dict[str, Any]],
    query_site_id: int,
) -> Dict[int, Dict[str, Any]]:
    """模拟 DWS 任务通过事实表反查 dim_member 的逻辑。

    等价于：
        SELECT * FROM dim_member
        WHERE member_id IN (
            SELECT DISTINCT tenant_member_id FROM fact_table
            WHERE site_id = :query_site_id
              AND tenant_member_id IS NOT NULL AND tenant_member_id != 0
        ) AND scd2_is_current = 1
    """
    # 步骤 1：从事实表中提取目标门店的 member_id 集合
    member_ids_in_fact: Set[int] = set()
    for row in fact_rows:
        if (
            row.get("site_id") == query_site_id
            and row.get("tenant_member_id") is not None
            and row.get("tenant_member_id") != 0
        ):
            member_ids_in_fact.add(row["tenant_member_id"])

    # 步骤 2：从 dim_member 中筛选当前有效版本
    result: Dict[int, Dict[str, Any]] = {}
    for m in dim_members:
        if m.get("scd2_is_current") == 1 and m["member_id"] in member_ids_in_fact:
            result[m["member_id"]] = dict(m)

    return result


def simulate_register_site_lookup(
    dim_members: List[Dict[str, Any]],
    query_site_id: int,
) -> Dict[int, Dict[str, Any]]:
    """模拟旧的 WHERE register_site_id = %s 查询模式（用于对比）。"""
    result: Dict[int, Dict[str, Any]] = {}
    for m in dim_members:
        if m.get("scd2_is_current") == 1 and m.get("register_site_id") == query_site_id:
            result[m["member_id"]] = dict(m)
    return result


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

# 门店 ID：2~5 个门店
_site_id_st = st.integers(min_value=1, max_value=5)

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

# 昵称 / 手机号
_nickname_st = st.text(
    alphabet=st.sampled_from("张李王赵刘陈杨黄周吴"),
    min_size=2,
    max_size=4,
)
_mobile_st = st.from_regex(r"1[3-9]\d{9}", fullmatch=True)


# 生成 dim_member 行
@st.composite
def _dim_member_st(draw):
    mid = draw(_member_id_st)
    return {
        "member_id": mid,
        "register_site_id": draw(_site_id_st),
        "nickname": draw(_nickname_st),
        "mobile": draw(_mobile_st),
        "scd2_is_current": 1,  # 只生成当前有效版本
    }


# 生成事实表行（会员在某门店的消费记录）
@st.composite
def _fact_row_st(draw, member_ids):
    """生成一条事实表记录，member_id 从已有会员中选取。"""
    mid = draw(st.sampled_from(member_ids)) if member_ids else draw(_member_id_st)
    return {
        "site_id": draw(_site_id_st),
        "tenant_member_id": mid,
    }


# 生成完整的跨店场景
@st.composite
def _cross_store_scenario_st(draw):
    """生成一个包含跨店消费会员的完整场景。

    保证至少有一个会员在非注册门店有消费记录。
    """
    # 生成 2~10 个会员
    members = draw(st.lists(_dim_member_st(), min_size=2, max_size=10, unique_by=lambda m: m["member_id"]))
    if not members:
        return None

    member_ids = [m["member_id"] for m in members]

    # 生成事实表记录（5~30 条）
    fact_rows = draw(st.lists(
        _fact_row_st(member_ids),
        min_size=5,
        max_size=30,
    ))

    # 强制注入至少一条跨店消费记录：
    # 选一个会员，在非注册门店生成消费
    cross_member = draw(st.sampled_from(members))
    other_sites = [s for s in range(1, 6) if s != cross_member["register_site_id"]]
    if other_sites:
        cross_site = draw(st.sampled_from(other_sites))
        fact_rows.append({
            "site_id": cross_site,
            "tenant_member_id": cross_member["member_id"],
        })
    else:
        cross_site = cross_member["register_site_id"]

    query_site_id = cross_site

    return {
        "members": members,
        "fact_rows": fact_rows,
        "query_site_id": query_site_id,
        "cross_member_id": cross_member["member_id"],
        "cross_member_register_site": cross_member["register_site_id"],
    }


# ── Property 6: 跨店会员可查 ─────────────────────────────────
# Feature: etl-aggregation-fix, Property 6: 跨店会员可查
# **Validates: Requirements 3.1, 3.2**
#
# 对于任意在 A 店注册但在 B 店有消费记录的会员，
# B 店的 DWS 任务通过事实表反查 dim_member 时，
# 应能获取到该会员的维度信息（nickname、mobile 等）。


class TestProperty6CrossStoreMemberVisible:
    """Property 6: 跨店会员可查。"""

    @given(scenario=_cross_store_scenario_st())
    @settings(max_examples=200)
    def test_cross_store_member_found_via_fact_lookup(self, scenario):
        """在 A 店注册、B 店消费的会员，B 店 DWS 通过事实表反查应能找到。"""
        if scenario is None:
            return

        members = scenario["members"]
        fact_rows = scenario["fact_rows"]
        query_site_id = scenario["query_site_id"]
        cross_member_id = scenario["cross_member_id"]

        result = simulate_fact_table_lookup(fact_rows, members, query_site_id)

        # 核心断言：跨店会员必须可查
        assert cross_member_id in result, (
            f"会员 {cross_member_id}（注册于门店 {scenario['cross_member_register_site']}）"
            f"在门店 {query_site_id} 有消费记录，但事实表反查未找到"
        )

        # 维度信息完整性
        member_info = result[cross_member_id]
        assert "nickname" in member_info, "缺少 nickname 维度信息"
        assert "mobile" in member_info, "缺少 mobile 维度信息"

    @given(scenario=_cross_store_scenario_st())
    @settings(max_examples=200)
    def test_old_register_site_pattern_misses_cross_store(self, scenario):
        """旧模式 WHERE register_site_id = %s 会遗漏跨店消费会员（反向验证）。"""
        if scenario is None:
            return

        members = scenario["members"]
        query_site_id = scenario["query_site_id"]
        cross_member_id = scenario["cross_member_id"]
        cross_register_site = scenario["cross_member_register_site"]

        # 跨店会员的注册门店 != 查询门店
        if cross_register_site == query_site_id:
            return  # 非跨店场景，跳过

        old_result = simulate_register_site_lookup(members, query_site_id)

        # 旧模式应该找不到跨店会员
        assert cross_member_id not in old_result, (
            f"旧模式 register_site_id={query_site_id} 不应找到"
            f"注册于门店 {cross_register_site} 的会员 {cross_member_id}"
        )

    @given(scenario=_cross_store_scenario_st())
    @settings(max_examples=200)
    def test_fact_lookup_superset_of_register_lookup(self, scenario):
        """事实表反查的结果应是 register_site_id 筛选结果的超集。

        即：旧模式能找到的会员，新模式也一定能找到。
        """
        if scenario is None:
            return

        members = scenario["members"]
        fact_rows = scenario["fact_rows"]
        query_site_id = scenario["query_site_id"]

        new_result = simulate_fact_table_lookup(fact_rows, members, query_site_id)
        old_result = simulate_register_site_lookup(members, query_site_id)

        # 旧模式找到的、且在事实表中有消费记录的会员，新模式也应找到
        fact_member_ids = {
            r["tenant_member_id"]
            for r in fact_rows
            if r["site_id"] == query_site_id
            and r["tenant_member_id"] is not None
            and r["tenant_member_id"] != 0
        }
        for mid in old_result:
            if mid in fact_member_ids:
                assert mid in new_result, (
                    f"会员 {mid} 在旧模式中可查且在事实表中有记录，"
                    f"但新模式未返回"
                )

    @given(
        members=st.lists(_dim_member_st(), min_size=1, max_size=10, unique_by=lambda m: m["member_id"]),
        query_site_id=_site_id_st,
    )
    @settings(max_examples=100)
    def test_no_fact_records_returns_empty(self, members, query_site_id):
        """事实表中无该门店消费记录时，反查结果应为空。"""
        # 空事实表
        result = simulate_fact_table_lookup([], members, query_site_id)
        assert len(result) == 0, (
            f"事实表为空时，门店 {query_site_id} 的反查结果应为空，"
            f"实际返回 {len(result)} 条"
        )

    @given(scenario=_cross_store_scenario_st())
    @settings(max_examples=100)
    def test_null_and_zero_member_ids_excluded(self, scenario):
        """事实表中 tenant_member_id 为 NULL 或 0 的记录不应参与反查。"""
        if scenario is None:
            return

        members = scenario["members"]
        query_site_id = scenario["query_site_id"]

        # 构造仅包含 NULL 和 0 的事实表
        bad_fact_rows = [
            {"site_id": query_site_id, "tenant_member_id": None},
            {"site_id": query_site_id, "tenant_member_id": 0},
        ]

        result = simulate_fact_table_lookup(bad_fact_rows, members, query_site_id)
        assert len(result) == 0, (
            f"tenant_member_id 为 NULL/0 时不应返回任何会员，"
            f"实际返回 {len(result)} 条"
        )
