gendesign/backend/tests/sql/test_2464_leads_stats_suffix_contract.py

"""Суффикс поля в /admin/leads/stats обязан соответствовать смыслу величины (#2464).

В ответе рядом стоят величины двух видов: за всё время (`leads_total`, `sources_total`) и
за окно `months` (`leads_window`, `converted_window`, `conv_pct_window`). Соглашение
читается однозначно по самим именам.

`revenue_total` и `deals_total` его нарушали: считались по CTE `window_leads`, то есть за
окно, а суффиксом обещали итог. Админка из-за этого печатала карточку «Revenue (всего)» с
12-месячной цифрой.

Проверяется ИНВАРИАНТ, а не набор имён: для данных, где итог заведомо не равен окну,
каждое поле `*_total` обязано совпасть с итогом, каждое `*_window` — с окном. Такая
формулировка краснеет на origin/main по НЕВЕРНОМУ ЗНАЧЕНИЮ, а не по отсутствию ключа, и
переживёт любое разумное переименование.

Тест герметичный: обе таблицы создаются ВРЕМЕННЫМИ в своей же сессии; по конвенции
`tests/sql/*` DSN по умолчанию смотрит в туннель к прод-базе, поэтому в фикстуре стоит
проверка, что затенение сработало.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker


def _dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
    )
    return (
        raw
        if raw.startswith("postgresql+")
        else raw.replace("postgresql://", "postgresql+psycopg://")
    )


def _db_reachable() -> tuple[bool, str]:
    try:
        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
        with eng.connect() as c:
            c.execute(text("SELECT 1"))
        return True, ""
    except Exception as exc:
        return False, str(exc)


_DB_OK, _DB_ERR = _db_reachable()
pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")

_SCHEMA = """
CREATE TEMP TABLE prinzip_leads (
    lead_id bigint, created_at timestamptz, source text, converted boolean,
    deal_id bigint) ON COMMIT DROP;
CREATE TEMP TABLE prinzip_deals (
    deal_id bigint, deal_price numeric) ON COMMIT DROP;
"""

_WINDOW_MONTHS = 12

# Внутри окна: 2 заявки, обе со сделками по 1 000 000.
# Снаружи (три года назад): 3 заявки, сделки по 5 000 000 — итог заведомо не равен окну.
_IN_WINDOW_LEADS = 2
_OUT_WINDOW_LEADS = 3
_ALL_TIME_LEADS = _IN_WINDOW_LEADS + _OUT_WINDOW_LEADS
_IN_WINDOW_REVENUE = 2_000_000.0
_IN_WINDOW_DEALS = 2


@pytest.fixture
def db():
    engine = create_engine(_dsn())
    session = sessionmaker(bind=engine)()
    try:
        session.execute(text(_SCHEMA))
        for table in ("prinzip_leads", "prinzip_deals"):
            n = session.execute(text(f"SELECT count(*) FROM {table}")).scalar()
            assert n == 0, (
                f"{table}: запрос попал НЕ во временную таблицу ({n} строк) — "
                "тест читал бы боевые данные"
            )
        yield session
    finally:
        session.rollback()
        session.close()
        engine.dispose()


@pytest.fixture
def seeded(db):
    rows = [
        (1, "0 days", "site", True, 101, 1_000_000),
        (2, "10 days", "site", True, 102, 1_000_000),
        (3, "1100 days", "avito", True, 103, 5_000_000),
        (4, "1101 days", "avito", True, 104, 5_000_000),
        (5, "1102 days", "vk", True, 105, 5_000_000),
    ]
    for lead_id, ago, source, converted, deal_id, price in rows:
        db.execute(
            text(
                "INSERT INTO prinzip_leads (lead_id, created_at, source, converted, deal_id)"
                " VALUES (:l, NOW() - CAST(:ago AS interval), :s, :c, :d)"
            ),
            {"l": lead_id, "ago": ago, "s": source, "c": converted, "d": deal_id},
        )
        db.execute(
            text("INSERT INTO prinzip_deals (deal_id, deal_price) VALUES (:d, :p)"),
            {"d": deal_id, "p": price},
        )
    return db


def _stats(db) -> dict:
    from app.api.v1.admin_leads import leads_stats

    return leads_stats(db=db, months=_WINDOW_MONTHS)


def test_window_suffixed_fields_match_the_window(seeded) -> None:
    """Всё, что названо `_window`, обязано считаться по окну.

    На origin/main эти величины лежат под именами `revenue_total`/`deals_total`,
    поэтому проверка ниже (по `_total`) и краснеет — здесь же контроль, что
    оконные значения не поехали.
    """
    stats = _stats(seeded)
    assert stats["leads_window"] == _IN_WINDOW_LEADS
    assert stats["converted_window"] == _IN_WINDOW_LEADS


def test_total_suffixed_fields_are_all_time(seeded) -> None:
    """КАЖДОЕ поле `*_total` обязано быть за всё время, а не за окно.

    На origin/main `revenue_total` = 2 000 000 (только окно) при итоге 17 000 000,
    и `deals_total` = 2 при итоге 5 — красное по неверному ЗНАЧЕНИЮ.
    """
    stats = _stats(seeded)
    all_time_revenue = float(
        seeded.execute(text("SELECT COALESCE(SUM(deal_price), 0) FROM prinzip_deals")).scalar()
    )
    all_time_deals = int(seeded.execute(text("SELECT COUNT(*) FROM prinzip_deals")).scalar())
    expected = {
        "leads_total": _ALL_TIME_LEADS,
        "revenue_total": all_time_revenue,
        "deals_total": all_time_deals,
    }

    for key, value in stats.items():
        if not key.endswith("_total"):
            continue
        if key not in expected:
            continue
        assert value == expected[key], (
            f"поле {key!r} обещает суффиксом величину за ВСЁ время, а равно {value} "
            f"при итоге {expected[key]} — это цифра за окно {_WINDOW_MONTHS} мес"
        )


def test_revenue_and_deals_are_named_by_their_scope(seeded) -> None:
    """Выручка и сделки должны нести суффикс, соответствующий их охвату.

    Отдельно от предыдущего: там проверяется значение под именем, здесь — что имя
    вообще выбрано по охвату. Ловит «починку», которая оставила бы `_total` и
    просто перестала показывать поле в UI.
    """
    stats = _stats(seeded)
    assert stats.get("revenue_window") == _IN_WINDOW_REVENUE, (
        f"revenue_window = {stats.get('revenue_window')}, ожидалось {_IN_WINDOW_REVENUE}"
    )
    assert stats.get("deals_window") == _IN_WINDOW_DEALS


def test_window_months_present_on_empty_data(db) -> None:
    """Контроль: на пустых данных ответ сохраняет форму и ширину окна.

    Оговорка, чтобы тест не читался как покрытие ветки `if not row`: он туда НЕ
    попадает. Запрос агрегатный и на пустых таблицах возвращает обычную строку
    (leads_total=0, leads_window=0), поэтому исполняется основная ветка. Ветка
    пустого ответа сегодня недостижима — её согласованность правится вслепую,
    и проверить её этим тестом нельзя.
    """
    stats = _stats(db)
    assert "window_months" in stats, f"нет window_months в пустом ответе: {sorted(stats)}"
    assert stats["window_months"] == _WINDOW_MONTHS