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

"""SQLAlchemy repository tests against an in-memory SQLite database.

Verifies day routing (latest dt + specific dt + missing) and rolling-row column
mapping against a real engine, with no Postgres needed.
"""

from __future__ import annotations

import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.db.models import (
    AdsTrdGroupFunnelDaily,
    AdsTrdGroupFunnelRolling,
    Base,
)
from app.schemas import DataStatus, Period
from app.services.funnel import run_funnel_query
from app.services.repository import SqlAlchemyFunnelRepository


def _daily_row(dt: str, top: int) -> AdsTrdGroupFunnelDaily:
    return AdsTrdGroupFunnelDaily(
        dt=dt,
        uv_start=top,
        uv_show=int(top * 0.8),
        uv_detail=int(top * 0.5),
        uv_order=int(top * 0.2),
        uv_paid=int(top * 0.16),
    )


def _rolling_row(dt: str, top7: int, top30: int) -> AdsTrdGroupFunnelRolling:
    return AdsTrdGroupFunnelRolling(
        dt=dt,
        uv_start_7d=top7,
        uv_show_7d=int(top7 * 0.8),
        uv_detail_7d=int(top7 * 0.5),
        uv_order_7d=int(top7 * 0.2),
        uv_paid_7d=int(top7 * 0.16),
        uv_start_30d=top30,
        uv_show_30d=int(top30 * 0.8),
        uv_detail_30d=int(top30 * 0.5),
        uv_order_30d=int(top30 * 0.2),
        uv_paid_30d=int(top30 * 0.16),
    )


@pytest.fixture
async def sqlite_sessionmaker():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    maker = async_sessionmaker(engine, expire_on_commit=False)
    yield maker
    await engine.dispose()


async def test_empty_daily_returns_none(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        assert await repo.fetch_daily(None) is None
        assert await repo.fetch_daily("20260623") is None


async def test_empty_rolling_returns_none(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        assert await repo.fetch_rolling() is None


async def test_daily_latest_dt_is_selected(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        session.add_all(
            [
                _daily_row("20260621", 100),
                _daily_row("20260623", 300),
                _daily_row("20260622", 200),
            ]
        )
        await session.commit()

    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        snapshot = await repo.fetch_daily(None)
        assert snapshot is not None
        assert snapshot.dt == "20260623"
        assert snapshot.values["uv_start"] == 300


async def test_daily_specific_dt(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        session.add_all([_daily_row("20260621", 100), _daily_row("20260623", 300)])
        await session.commit()

    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        snapshot = await repo.fetch_daily("20260621")
        assert snapshot is not None
        assert snapshot.dt == "20260621"
        assert snapshot.values["uv_start"] == 100


async def test_daily_specific_dt_missing(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        session.add(_daily_row("20260623", 300))
        await session.commit()

    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        assert await repo.fetch_daily("20200101") is None


async def test_rolling_row_columns(sqlite_sessionmaker) -> None:
    async with sqlite_sessionmaker() as session:
        session.add(_rolling_row("20260623", top7=700, top30=3000))
        await session.commit()

    async with sqlite_sessionmaker() as session:
        repo = SqlAlchemyFunnelRepository(session)
        resp7 = await run_funnel_query(Period.last_7d, repo)
        assert resp7.data_status == DataStatus.ready
        assert resp7.snapshot_dt == "20260623"
        assert resp7.results[0].uv == 700

        resp30 = await run_funnel_query(Period.last_30d, repo)
        assert resp30.results[0].uv == 3000