"""按月分块的**机制**回归 + 读侧响应缓存（2026-07-29 复审）。

分块的等价性测试（`test_cost_dual_measure_read_side` 里那条）在未分块
实现上同样通过 —— 它断言的每条性质（跨月合计、全窗口 top-N、other 桶）
都是「分块与否都成立」的，所以整套测试里没有一条能发现分块被拆掉。这里
钉的是机制本身：用 SQLAlchemy 的 ``before_cursor_execute`` 数真实发出的
SELECT 条数，跨月窗口必须逐月发查询，而不是一条宽 GROUP BY（TiDB
serverless 单 SQL 内存配额 8175：2026-07-28 生产事故，一次扫 6 个月
~550 万行）。

同时钉：
  * ``detect_anomalies``（本文件最宽的 GROUP BY:integration × resource
    × date）与 ``subordinates_cost_breakdown`` 也按月分块，且分块结果与
    未分块口径一致 —— 含 ``detect_anomalies`` 的整组口径闸必须是**全
    窗口**语义（分块内判定会让「回填边界横跨月边界」的资源假通过）。
  * ``/breakdown`` / ``/timeseries`` 的 Redis 响应缓存，键必须按口径隔离
    （保密红线:一个角色绝不能吃到另一个口径的缓存体）。
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest
import pytest_asyncio
from sqlalchemy import event, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.cost_service import (
    breakdown,
    detect_anomalies,
    subordinates_cost_breakdown,
    timeseries,
)
from app.domains.cmdb.models import CmdbCostFact
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


@pytest_asyncio.fixture
async def chunk_integration(db_session: AsyncSession) -> Integration:
    integ = Integration(
        id=str(uuid.uuid4()), kind="cloud", provider="aws",
        name="chunk-exec", slug=f"chunk-{uuid.uuid4().hex[:8]}", enabled=True,
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


async def _seed(
    db: AsyncSession, integ: Integration, *, day: date, service: str,
    invoice: str, list_amt: str | None = None,
    resource_id: str | None = None, person_id: uuid.UUID | None = None,
    department: str | None = None,
) -> None:
    db.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex, date=day, integration_id=str(integ.id),
        service=service, resource_id=resource_id,
        amount_usd=Decimal(invoice), amount_native=Decimal(invoice),
        amount_usd_list=(
            Decimal(list_amt) if list_amt is not None else None
        ),
        amount_native_list=(
            Decimal(list_amt) if list_amt is not None else None
        ),
        cost_owner_person_id=str(person_id) if person_id else None,
        cost_owner_department=department,
        cost_bucket="usage", currency="USD",
    ))
    await db.commit()


class _SelectCounter:
    """统计某个 engine 上真实发出的 SELECT 语句条数。"""

    def __init__(self, session: AsyncSession) -> None:
        self._engine = session.bind.sync_engine  # type: ignore[union-attr]
        self.statements: list[str] = []

    def __enter__(self) -> _SelectCounter:
        event.listen(self._engine, "before_cursor_execute", self._on)
        return self

    def __exit__(self, *_exc) -> None:
        event.remove(self._engine, "before_cursor_execute", self._on)

    def _on(self, _conn, _cursor, statement, _params, _ctx, _many) -> None:
        if statement.strip().upper().startswith("SELECT"):
            self.statements.append(statement)

    def matching(self, needle: str) -> list[str]:
        return [s for s in self.statements if needle in s]


@pytest.mark.asyncio
async def test_breakdown_issues_one_query_set_per_month(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    """3 个自然月的窗口 → 每月一组聚合（而不是一条跨 3 个月的宽
    GROUP BY）。等价性测试对分块与否无感,这条才是 8175 防线的锚。"""
    integ_id = str(chunk_integration.id)
    await _seed(
        db_session, chunk_integration, day=date(2026, 5, 10),
        service="EC2", invoice="10", list_amt="10",
    )

    def _grouped(counter: _SelectCounter) -> list[str]:
        # 分组聚合语句（不含 SP / 未分摊 / 未换算三个标量子查询）。
        return [
            s for s in counter.statements
            if "GROUP BY" in s.upper()
            and "cmdb_cost_facts" in s
            and "auto_owner_source =" not in s
        ]

    with _SelectCounter(db_session) as three_months:
        await breakdown(
            db_session, start=date(2026, 5, 1), end=date(2026, 7, 31),
            group_by="service", integration_id=integ_id, top=10,
            measure="list",
        )
    with _SelectCounter(db_session) as one_month:
        await breakdown(
            db_session, start=date(2026, 5, 1), end=date(2026, 5, 31),
            group_by="service", integration_id=integ_id, top=10,
            measure="list",
        )
    n3 = len(_grouped(three_months))
    n1 = len(_grouped(one_month))
    assert n1 >= 1, one_month.statements
    assert n3 == 3 * n1, (n3, n1)
    # 单条 SQL 的日期窗口不得跨月:每条聚合都带两个日期边界参数,
    # 且总条数随月份数线性增长（未分块实现恒等于 n1）。
    assert len(three_months.statements) == 3 * len(one_month.statements)


@pytest.mark.asyncio
async def test_timeseries_issues_one_query_set_per_month(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    integ_id = str(chunk_integration.id)
    await _seed(
        db_session, chunk_integration, day=date(2026, 6, 10),
        service="EC2", invoice="10", list_amt="10",
    )
    with _SelectCounter(db_session) as two_months:
        await timeseries(
            db_session, start=date(2026, 6, 1), end=date(2026, 7, 31),
            group_by="service", integration_id=integ_id, top=5,
            measure="list",
        )
    with _SelectCounter(db_session) as one_month:
        await timeseries(
            db_session, start=date(2026, 6, 1), end=date(2026, 6, 30),
            group_by="service", integration_id=integ_id, top=5,
            measure="list",
        )
    assert len(one_month.statements) >= 3, one_month.statements
    assert len(two_months.statements) == 2 * len(one_month.statements)


@pytest.mark.asyncio
async def test_subordinates_breakdown_chunks_and_matches_unchunked(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    """365 天窗口的四个聚合同样按月分块;top-20 的截断在合并之后 ——
    否则某个单月冠军会挤掉全窗口第一。"""
    person = IdentityPerson(
        id=uuid.uuid4(), source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Chunk Owner",
        work_email=f"chunk-{uuid.uuid4().hex[:6]}@x.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.commit()

    # Alpha 全窗口 30(每月 10),Gamma 只有 7 月的 20 —— 单月冠军。
    for day in (date(2026, 5, 10), date(2026, 6, 10), date(2026, 7, 10)):
        await _seed(
            db_session, chunk_integration, day=day, service="Alpha",
            invoice="10", list_amt="10", person_id=person.id,
            department="D1",
        )
    await _seed(
        db_session, chunk_integration, day=date(2026, 7, 20),
        service="Gamma", invoice="20", list_amt="20", person_id=person.id,
        department="D1",
    )

    with _SelectCounter(db_session) as counter:
        out = await subordinates_cost_breakdown(
            db_session, person_ids={person.id},
            start=date(2026, 5, 1), end=date(2026, 7, 31),
            integration_id=str(chunk_integration.id), measure="list",
        )
    fact_queries = [
        s for s in counter.statements if "cmdb_cost_facts" in s
    ]
    # 四个聚合 × 3 个月 = 12 条 facts 查询（未分块实现只有 4 条）。
    assert len(fact_queries) == 12, len(fact_queries)
    for s in fact_queries:
        assert "cmdb_cost_facts.date >=" in s and "date <=" in s

    assert out["total_usd"] == pytest.approx(50.0)
    assert out["fact_count"] == 4
    by_service = {r["service"]: r["cost_usd"] for r in out["by_service"]}
    assert by_service == {
        "Alpha": pytest.approx(30.0), "Gamma": pytest.approx(20.0),
    }
    # 排序按全窗口合计:Alpha(30) 在 Gamma(20) 前。
    assert [r["service"] for r in out["by_service"]] == ["Alpha", "Gamma"]
    assert out["by_person"][0]["cost_usd"] == pytest.approx(50.0)
    assert out["by_department"][0]["department"] == "D1"
    assert out["by_department"][0]["cost_usd"] == pytest.approx(50.0)


@pytest.mark.asyncio
async def test_detect_anomalies_chunks_and_keeps_whole_window_measure_gate(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    """最宽的 GROUP BY 也按月分块,且整组口径闸仍是**全窗口**语义:
    「早期天缺 list、后期天有 list」的资源横跨月边界时,必须整组回落
    净额（分块内判定会让后一个月的分块单独通过 list 闸,重现 1.65×
    假尖峰的混合基线）。"""
    anchor = datetime.now(UTC).date() - timedelta(days=1)
    start = anchor - timedelta(days=39)  # 40 天窗口,必然跨月
    integ = chunk_integration

    # 资源 A:窗口前半缺 list、后半每天都有 list,尖峰在最后一天的
    # **list** 列上;净额恒 100(无尖峰)。整组回落净额 → 不报。
    # 若把口径闸放进分块,后一个月的分块「每行都有 list」会单独通过
    # list 闸,于是用 list 序列(100…300)报出假尖峰。
    for i in range(40):
        day = start + timedelta(days=i)
        await _seed(
            db_session, integ, day=day, service="EC2", invoice="100",
            list_amt=(
                None if i < 20 else ("300" if i == 39 else "100")
            ),
            resource_id="i-straddle",
        )

    with _SelectCounter(db_session) as counter:
        out = await detect_anomalies(
            db_session, days=40, integration_id=str(integ.id),
            min_impact_usd=50.0, measure="list",
        )
    fact_queries = [s for s in counter.statements if "cmdb_cost_facts" in s]
    n_months = len({
        (start + timedelta(days=i)).strftime("%Y-%m") for i in range(40)
    })
    assert n_months >= 2
    assert len(fact_queries) == n_months, (len(fact_queries), n_months)

    by_rid = {row["resource_id"] for row in out}
    # 全窗口整组闸:混合覆盖 → 整组净额(恒 100)→ 绝无假尖峰。
    assert "i-straddle" not in by_rid, out

    # 对照:整窗口每天都有 list 的资源走 list 序列。
    for i in range(40):
        day = start + timedelta(days=i)
        amt = "300" if i == 39 else "100"
        await _seed(
            db_session, integ, day=day, service="EC2", invoice="1",
            list_amt=amt, resource_id="i-all-list",
        )
    out2 = await detect_anomalies(
        db_session, days=40, integration_id=str(integ.id),
        min_impact_usd=50.0, measure="list",
    )
    row2 = next(r for r in out2 if r["resource_id"] == "i-all-list")
    assert row2["amount_usd"] == pytest.approx(300.0)


@pytest.mark.asyncio
async def test_detect_anomalies_can_use_shorter_chunks(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    """A high-cardinality all-account caller can cap each GROUP BY window."""
    anchor = datetime.now(UTC).date() - timedelta(days=1)
    for i in range(14):
        await _seed(
            db_session, chunk_integration, day=anchor - timedelta(days=i),
            service="EC2", invoice="100", list_amt="100", resource_id="i-short-chunks",
        )

    with _SelectCounter(db_session) as counter:
        await detect_anomalies(
            db_session, days=14, integration_id=str(chunk_integration.id),
            measure="list", chunk_days=7,
        )

    fact_queries = [s for s in counter.statements if "cmdb_cost_facts" in s]
    assert len(fact_queries) == 2


@pytest.mark.asyncio
async def test_breakdown_decimal_accumulation_across_chunks(
    db_session: AsyncSession, chunk_integration: Integration,
) -> None:
    """跨月合并用 Decimal 累加:12 个月 × 0.07 必须精确等于 0.84。
    float 累加会给出 0.8400000000000001 —— round(…, 2) 恰好把它藏住,
    但 total/other/share 的链式运算会把误差抬到分位上,所以这里同时
    断言分组值、合计与 other 桶。"""
    integ_id = str(chunk_integration.id)
    start = date(2026, 1, 1)
    for m in range(1, 13):
        await _seed(
            db_session, chunk_integration, day=date(2026, m, 15),
            service="Cent", invoice="0.07", list_amt="0.07",
        )
        await _seed(
            db_session, chunk_integration, day=date(2026, m, 16),
            service="Other", invoice="0.01", list_amt="0.01",
        )
    out = await breakdown(
        db_session, start=start, end=date(2026, 12, 31),
        group_by="service", integration_id=integ_id, top=1, measure="list",
    )
    assert out["groups"][0]["key"] == "Cent"
    assert out["groups"][0]["cost_usd"] == 0.84
    assert out["other_usd"] == 0.12
    assert out["total_usd"] == 0.96


@pytest.mark.asyncio
async def test_breakdown_and_timeseries_cache_is_measure_isolated(
    client, finance_headers, db_session: AsyncSession,
    chunk_integration: Integration,
) -> None:
    """响应缓存契约（与财务瀑布同款）+ 保密红线:缓存键含口径,
    ``measure=invoice`` 绝不能吃到 ``measure=list`` 的缓存体。"""
    from app.core.config import settings as _settings
    from app.core.redis_client import get_redis

    integ_id = str(chunk_integration.id)
    day = datetime.now(UTC).date() - timedelta(days=2)
    await _seed(
        db_session, chunk_integration, day=day, service="EC2",
        invoice="1", list_amt="100",
    )
    params = {
        "group_by": "service", "integration_id": integ_id,
        "start": day.isoformat(), "end": day.isoformat(),
    }
    prev = _settings.COST_READ_CACHE_TTL_SECONDS
    _settings.COST_READ_CACHE_TTL_SECONDS = 60
    keys_seen: list[str] = []
    try:
        r_list = await client.get(
            "/api/v1/cmdb/cost/breakdown", headers=finance_headers,
            params=params,
        )
        assert r_list.status_code == 200, r_list.text
        assert r_list.json()["data"]["total_usd"] == pytest.approx(100.0)

        # 口径隔离:同窗口同参数,invoice 必须现算(1.0),不吃 list 的
        # 缓存体(100.0)。
        r_inv = await client.get(
            "/api/v1/cmdb/cost/breakdown", headers=finance_headers,
            params={**params, "measure": "invoice"},
        )
        assert r_inv.status_code == 200, r_inv.text
        assert r_inv.json()["data"]["measure"] == "invoice"
        assert r_inv.json()["data"]["total_usd"] == pytest.approx(1.0)

        # 缓存生效:命中窗口内新增的行看不见。
        await _seed(
            db_session, chunk_integration, day=day, service="S3",
            invoice="5", list_amt="50",
        )
        r_again = await client.get(
            "/api/v1/cmdb/cost/breakdown", headers=finance_headers,
            params=params,
        )
        assert r_again.json()["data"]["total_usd"] == pytest.approx(100.0)

        # timeseries 同样走缓存 + 口径隔离。
        r_ts = await client.get(
            "/api/v1/cmdb/cost/timeseries", headers=finance_headers,
            params=params,
        )
        assert r_ts.status_code == 200, r_ts.text
        assert sum(r_ts.json()["data"]["total_per_day"]) == pytest.approx(
            150.0,
        )
        r_ts_inv = await client.get(
            "/api/v1/cmdb/cost/timeseries", headers=finance_headers,
            params={**params, "measure": "invoice"},
        )
        assert sum(
            r_ts_inv.json()["data"]["total_per_day"],
        ) == pytest.approx(6.0)

        redis = get_redis()
        async for key in redis.scan_iter("cost:breakdown:v1:*"):
            keys_seen.append(
                key if isinstance(key, str) else key.decode(),
            )
        # 两个口径两把键 —— 键里带口径,不是同一把。
        assert len({k.split(":")[3] for k in keys_seen}) == 2, keys_seen
    finally:
        _settings.COST_READ_CACHE_TTL_SECONDS = prev
        redis = get_redis()
        for pattern in ("cost:breakdown:v1:*", "cost:timeseries:v1:*"):
            async for key in redis.scan_iter(pattern):
                await redis.delete(key)


@pytest.mark.asyncio
async def test_breakdown_endpoint_fans_out_over_separate_sessions(
    client, finance_headers, db_session: AsyncSession,
    chunk_integration: Integration,
) -> None:
    """HTTP 路径把 ``async_session_factory`` 交给分块（有界并发,独立
    session —— 同一 AsyncSession 不能并发）。跨月窗口的合计必须与直调
    服务层一致,证明并发分块读到的是同一份数据。"""
    integ_id = str(chunk_integration.id)
    today = datetime.now(UTC).date()
    cur_month = today.replace(day=1)
    prev_month = (cur_month - timedelta(days=1)).replace(day=1)
    await _seed(
        db_session, chunk_integration, day=prev_month, service="EC2",
        invoice="30", list_amt="50",
    )
    await _seed(
        db_session, chunk_integration, day=cur_month, service="EC2",
        invoice="70", list_amt="110",
    )
    # 计数 session 的开启次数:分块数 == 独立 session 数(顺序执行会是 0)。
    from app.core import database as _db

    factory = _db.async_session_factory
    opened = {"n": 0}

    def _counting_factory(*a, **kw):
        opened["n"] += 1
        return factory(*a, **kw)

    _db.async_session_factory = _counting_factory
    try:
        r = await client.get(
            "/api/v1/cmdb/cost/breakdown", headers=finance_headers,
            params={
                "group_by": "service", "integration_id": integ_id,
                "start": prev_month.isoformat(), "end": today.isoformat(),
            },
        )
    finally:
        _db.async_session_factory = factory
    assert r.status_code == 200, r.text
    assert r.json()["data"]["total_usd"] == pytest.approx(160.0)
    assert opened["n"] == 2, opened  # 两个自然月 → 两个独立 session
    direct = await breakdown(
        db_session, start=prev_month, end=today, group_by="service",
        integration_id=integ_id, top=20, measure="list",
    )
    assert direct["total_usd"] == pytest.approx(160.0)
    # 落库行数不变（只读路径,分块不写）。
    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.integration_id == integ_id,
        ),
    )).scalars().all()
    assert len(rows) == 2