hs-data/apps/api/tests/test_funnel_service.py

"""Tests for the group-buy funnel service: routing, conversion math,
data_status, snapshot_dt validation, and the fake data source."""

from __future__ import annotations

from datetime import date, timedelta

import pytest

from app.schemas import DataStatus, Period
from app.services.funnel import build_results, run_funnel_query
from app.services.repository import FakeFunnelRepository
from tests.conftest import FakeRepository


def _daily(uvs: list[int | None]) -> dict[str, int | None]:
    keys = ["start", "show", "detail", "order", "paid"]
    return {f"uv_{k}": uv for k, uv in zip(keys, uvs)}


def _rolling(
    uvs_7d: list[int | None] | None = None,
    uvs_30d: list[int | None] | None = None,
) -> dict[str, int | None]:
    keys = ["start", "show", "detail", "order", "paid"]
    uvs_7d = uvs_7d if uvs_7d is not None else [None] * 5
    uvs_30d = uvs_30d if uvs_30d is not None else [None] * 5
    values: dict[str, int | None] = {}
    for k, uv in zip(keys, uvs_7d):
        values[f"uv_{k}_7d"] = uv
    for k, uv in zip(keys, uvs_30d):
        values[f"uv_{k}_30d"] = uv
    return values


# --------------------------------------------------------------------------- #
# Conversion math                                                             #
# --------------------------------------------------------------------------- #


def test_build_results_step1_null_rates() -> None:
    results = build_results([10000, 8200, 5100, 2200, 1800])
    assert results[0].step_index == 1
    assert results[0].conversion_rate is None
    assert results[0].dropoff_rate is None


def test_build_results_descending_funnel() -> None:
    results = build_results([10000, 8200, 5100, 2200, 1800])
    assert [r.uv for r in results] == [10000, 8200, 5100, 2200, 1800]
    assert [r.event_key for r in results] == [
        "start",
        "show",
        "detail",
        "order",
        "paid",
    ]
    # Step 3 display name is "拼团详情" (was "详情").
    assert [r.name for r in results] == ["启动", "曝光", "拼团详情", "下单", "成功"]
    assert results[1].conversion_rate == pytest.approx(0.82)
    assert results[1].dropoff_rate == pytest.approx(0.18)
    assert results[2].conversion_rate == pytest.approx(5100 / 8200)
    assert results[3].conversion_rate == pytest.approx(2200 / 5100)


def test_build_results_prev_zero_yields_null() -> None:
    results = build_results([0, 5, 3, 1, 0])
    assert results[1].conversion_rate is None
    assert results[1].dropoff_rate is None
    assert results[2].conversion_rate == pytest.approx(0.6)


# --------------------------------------------------------------------------- #
# day routing                                                                 #
# --------------------------------------------------------------------------- #


async def test_day_latest_when_no_snapshot_dt(repo: FakeRepository) -> None:
    repo.set_daily("20260620", _daily([1, 1, 1, 1, 1]))
    repo.set_daily("20260623", _daily([100, 80, 50, 20, 10]))  # latest
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=None)
    assert resp.data_status == DataStatus.ready
    assert resp.snapshot_dt == "20260623"
    assert [r.uv for r in resp.results] == [100, 80, 50, 20, 10]


async def test_day_specific_dt(repo: FakeRepository) -> None:
    yesterday = date.today() - timedelta(days=1)
    older = yesterday - timedelta(days=3)
    repo.set_daily(yesterday.strftime("%Y%m%d"), _daily([100, 80, 50, 20, 10]))
    repo.set_daily(older.strftime("%Y%m%d"), _daily([90, 70, 40, 15, 8]))
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=older)
    assert resp.snapshot_dt == older.strftime("%Y%m%d")
    assert [r.uv for r in resp.results] == [90, 70, 40, 15, 8]


async def test_day_nonexistent_dt_is_missing(repo: FakeRepository) -> None:
    yesterday = date.today() - timedelta(days=1)
    repo.set_daily(yesterday.strftime("%Y%m%d"), _daily([100, 80, 50, 20, 10]))
    # A valid (<= yesterday) date with no row installed.
    missing_day = yesterday - timedelta(days=400)
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=missing_day)
    assert resp.data_status == DataStatus.missing
    assert resp.snapshot_dt is None
    assert resp.results == []


async def test_day_no_rows_at_all_is_missing(repo: FakeRepository) -> None:
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=None)
    assert resp.data_status == DataStatus.missing
    assert resp.snapshot_dt is None


async def test_day_null_column_is_missing(repo: FakeRepository) -> None:
    repo.set_daily("20260623", _daily([100, 80, None, 20, 10]))
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=None)
    assert resp.data_status == DataStatus.missing
    assert resp.snapshot_dt == "20260623"


async def test_day_future_snapshot_dt_raises(repo: FakeRepository) -> None:
    from app.services.funnel import SnapshotDateError

    with pytest.raises(SnapshotDateError):
        await run_funnel_query(
            Period.day, repo, snapshot_dt=date.today() + timedelta(days=1)
        )


async def test_day_today_snapshot_dt_raises(repo: FakeRepository) -> None:
    from app.services.funnel import SnapshotDateError

    with pytest.raises(SnapshotDateError):
        await run_funnel_query(Period.day, repo, snapshot_dt=date.today())


# --------------------------------------------------------------------------- #
# rolling routing                                                             #
# --------------------------------------------------------------------------- #


async def test_rolling_7d_picks_7d_columns(repo: FakeRepository) -> None:
    repo.set_rolling(
        "20260623",
        _rolling(
            uvs_7d=[700, 560, 350, 140, 70],
            uvs_30d=[3000, 2400, 1500, 600, 300],
        ),
    )
    resp = await run_funnel_query(Period.last_7d, repo)
    assert resp.data_status == DataStatus.ready
    assert resp.snapshot_dt == "20260623"
    assert [r.uv for r in resp.results] == [700, 560, 350, 140, 70]


async def test_rolling_30d_picks_30d_columns(repo: FakeRepository) -> None:
    repo.set_rolling(
        "20260623",
        _rolling(
            uvs_7d=[700, 560, 350, 140, 70],
            uvs_30d=[3000, 2400, 1500, 600, 300],
        ),
    )
    resp = await run_funnel_query(Period.last_30d, repo)
    assert [r.uv for r in resp.results] == [3000, 2400, 1500, 600, 300]


async def test_rolling_ignores_snapshot_dt(repo: FakeRepository) -> None:
    repo.set_rolling("20260623", _rolling(uvs_7d=[700, 560, 350, 140, 70]))
    # A future snapshot_dt would be rejected for period=day, but must be
    # silently ignored for rolling periods.
    resp = await run_funnel_query(
        Period.last_7d, repo, snapshot_dt=date.today() + timedelta(days=5)
    )
    assert resp.data_status == DataStatus.ready
    assert resp.snapshot_dt == "20260623"


async def test_rolling_missing_when_no_row(repo: FakeRepository) -> None:
    resp = await run_funnel_query(Period.last_7d, repo)
    assert resp.data_status == DataStatus.missing
    assert resp.snapshot_dt is None
    assert resp.results == []


async def test_rolling_missing_when_columns_null(repo: FakeRepository) -> None:
    # Row present, 30d columns NULL.
    repo.set_rolling("20260623", _rolling(uvs_7d=[700, 560, 350, 140, 70]))
    resp = await run_funnel_query(Period.last_30d, repo)
    assert resp.data_status == DataStatus.missing
    assert resp.snapshot_dt == "20260623"


# --------------------------------------------------------------------------- #
# Fake data source                                                            #
# --------------------------------------------------------------------------- #


async def test_fake_source_ready_for_all_periods() -> None:
    repo = FakeFunnelRepository()
    yesterday = date.today() - timedelta(days=1)
    for period in (Period.day, Period.last_7d, Period.last_30d):
        resp = await run_funnel_query(period, repo)
        assert resp.data_status == DataStatus.ready
        assert resp.snapshot_dt == yesterday.strftime("%Y%m%d")
        assert len(resp.results) == 5
        uvs = [r.uv for r in resp.results]
        assert uvs == sorted(uvs, reverse=True)
        assert resp.results[0].conversion_rate is None


async def test_fake_source_historical_day_ready() -> None:
    repo = FakeFunnelRepository()
    yesterday = date.today() - timedelta(days=1)
    older = yesterday - timedelta(days=4)
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=older)
    assert resp.data_status == DataStatus.ready
    assert resp.snapshot_dt == older.strftime("%Y%m%d")
    assert len(resp.results) == 5


async def test_fake_source_history_differs_by_day() -> None:
    repo = FakeFunnelRepository()
    yesterday = date.today() - timedelta(days=1)
    older = yesterday - timedelta(days=5)
    resp_latest = await run_funnel_query(Period.day, repo, snapshot_dt=yesterday)
    resp_older = await run_funnel_query(Period.day, repo, snapshot_dt=older)
    # Older day scaled down, so the top UVs differ.
    assert resp_latest.results[0].uv != resp_older.results[0].uv


async def test_fake_source_nonexistent_day_missing() -> None:
    repo = FakeFunnelRepository()
    yesterday = date.today() - timedelta(days=1)
    far_past = yesterday - timedelta(days=500)
    resp = await run_funnel_query(Period.day, repo, snapshot_dt=far_past)
    assert resp.data_status == DataStatus.missing