"""Tests for period validation, period -> column mapping, and snapshot_dt rules.""" from __future__ import annotations from datetime import date, timedelta import pytest from app.schemas import FunnelQueryRequest, Period from app.services.funnel import ( ROLLING_SUFFIX, SnapshotDateError, daily_columns, rolling_columns, validate_snapshot_dt, ) def test_period_accepts_three_valid_values() -> None: for value in ("day", "last_7d", "last_30d"): req = FunnelQueryRequest(period=value) assert req.period.value == value def test_period_rejects_bad_value() -> None: with pytest.raises(ValueError): FunnelQueryRequest(period="yesterday") # old v2 value, now invalid def test_request_snapshot_dt_optional() -> None: req = FunnelQueryRequest(period="day") assert req.snapshot_dt is None req2 = FunnelQueryRequest(period="day", snapshot_dt="2026-06-20") assert req2.snapshot_dt == date(2026, 6, 20) def test_rolling_suffix_mapping() -> None: assert ROLLING_SUFFIX[Period.last_7d] == "7d" assert ROLLING_SUFFIX[Period.last_30d] == "30d" def test_daily_columns() -> None: assert daily_columns() == [ "uv_start", "uv_show", "uv_detail", "uv_order", "uv_paid", ] def test_rolling_columns_last_7d() -> None: assert rolling_columns(Period.last_7d) == [ "uv_start_7d", "uv_show_7d", "uv_detail_7d", "uv_order_7d", "uv_paid_7d", ] def test_rolling_columns_last_30d() -> None: assert rolling_columns(Period.last_30d) == [ "uv_start_30d", "uv_show_30d", "uv_detail_30d", "uv_order_30d", "uv_paid_30d", ] def test_validate_snapshot_dt_accepts_yesterday() -> None: yesterday = date.today() - timedelta(days=1) validate_snapshot_dt(yesterday) # no raise def test_validate_snapshot_dt_accepts_old_date() -> None: validate_snapshot_dt(date.today() - timedelta(days=30)) # no raise def test_validate_snapshot_dt_rejects_today() -> None: with pytest.raises(SnapshotDateError): validate_snapshot_dt(date.today()) def test_validate_snapshot_dt_rejects_future() -> None: with pytest.raises(SnapshotDateError): validate_snapshot_dt(date.today() + timedelta(days=1))