gendesign/tradein-mvp/backend/tests/test_sell_time_sensitivity.py

"""Tests for GET /estimate/{id}/sell-time-sensitivity (#1995).

Regression coverage for the "недостаточно данных" honesty flag: buckets built
from a thin sample (n_lots below settings.sell_time_sensitivity_min_n_lots)
must carry insufficient_data=True instead of silently exposing noisy
median/p25/p75 exposure_days numbers. We deliberately do NOT try to smooth or
"fix" a non-monotonic result (e.g. +10% selling faster than +5%) — see
test_sell_time_sensitivity_does_not_smooth_non_monotonic_result below.
"""

from __future__ import annotations

import os
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock

# psycopg v3 driver required; stub DATABASE_URL before any app import
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import.
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
sys.modules.setdefault("weasyprint.HTML", _wp_mock)

import pytest  # noqa: E402
from fastapi import FastAPI  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

_ESTIMATE_ID = "22222222-2222-2222-2222-222222222222"


@pytest.fixture(autouse=True)
def _restore_get_role():
    """Restore app.core.auth.get_role after each test (mirror test_estimate_idor)."""
    from app.core import auth as auth_mod

    original = auth_mod.get_role
    yield
    auth_mod.get_role = original


def _make_app() -> FastAPI:
    """Minimal FastAPI app mounting only the trade-in router."""
    from app.api.v1 import trade_in as trade_in_module

    application = FastAPI()
    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
    return application


def _exec_result(
    *,
    fetchone: object | None = None,
    all_rows: list | None = None,
    scalar: object | None = None,
    mapping_rows: list | None = None,
) -> MagicMock:
    """Build a single db.execute(...) return value supporting whichever chain
    the endpoint calls next (.fetchone() / .all() / .scalar() / .mappings().all())."""
    result = MagicMock()
    result.fetchone.return_value = fetchone
    result.all.return_value = all_rows if all_rows is not None else []
    result.scalar.return_value = scalar
    mapping_mock = MagicMock()
    mapping_mock.all.return_value = mapping_rows if mapping_rows is not None else []
    result.mappings.return_value = mapping_mock
    return result


def _make_db_mock(
    *,
    created_by: str | None,
    bucket_rows: list[dict],
    target_median: int | None = 120_000,
) -> MagicMock:
    """Sequential db.execute() mock covering the endpoint's fixed 6-call path
    (explicit radius_m, address resolves ≥1 house_id):
      1. _assert_estimate_access_by_id → fetchone(created_by)
      2. target lat/lon/address → fetchone
      3. address-based house lookup → all()
      4. radius_m-based house expansion → all()
      5. target_median benchmark → scalar()
      6. bucket_rows → mappings().all()
    """
    db = MagicMock()
    db.execute.side_effect = [
        _exec_result(fetchone=SimpleNamespace(created_by=created_by)),
        _exec_result(
            fetchone=SimpleNamespace(lat=56.8, lon=60.6, address="Екатеринбург, ул. Тестовая, 1")
        ),
        _exec_result(all_rows=[SimpleNamespace(id=1)]),
        _exec_result(all_rows=[SimpleNamespace(id=2)]),
        _exec_result(scalar=target_median),
        _exec_result(mapping_rows=bucket_rows),
    ]
    return db


def _client_with(app: FastAPI, db_mock: MagicMock, role: str) -> TestClient:
    from app.core.db import get_db

    def _override_db():
        yield db_mock

    app.dependency_overrides[get_db] = _override_db

    auth_mod = sys.modules["app.core.auth"]
    auth_mod.get_role = lambda _u: role  # type: ignore[assignment]
    return TestClient(app)


def _bucket_row(bucket: str, n_lots: int, median_exposure_days: int) -> dict:
    return {
        "bucket": bucket,
        "n_lots": n_lots,
        "median_exposure_days": median_exposure_days,
        "p25_days": median_exposure_days - 5,
        "p75_days": median_exposure_days + 5,
    }


def _get(client: TestClient, *, radius_m: int = 500) -> dict:
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/sell-time-sensitivity",
        params={"radius_m": radius_m},
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200, resp.text
    return resp.json()


# ── Test: thin bucket (n_lots < threshold) flagged insufficient_data ────────


def test_sell_time_sensitivity_flags_thin_bucket_insufficient() -> None:
    """n_lots below settings.sell_time_sensitivity_min_n_lots (default 10) →
    insufficient_data=True; buckets with plenty of lots stay False."""
    from app.core.config import settings

    bucket_rows = [
        _bucket_row("cheap", 20, 30),
        _bucket_row("median", 18, 35),
        # plus5 — thin sample (3 lots).
        _bucket_row("plus5", 3, 60),
        _bucket_row("plus10", 15, 40),
    ]
    db_mock = _make_db_mock(created_by="kopylov", bucket_rows=bucket_rows)
    app = _make_app()
    client = _client_with(app, db_mock, role="admin")
    data = _get(client)

    by_label = {b["price_premium_label"]: b for b in data["buckets"]}
    assert by_label["cheap"]["n_lots"] == 20
    assert by_label["cheap"]["insufficient_data"] is False
    assert by_label["plus5"]["n_lots"] == 3
    assert by_label["plus5"]["insufficient_data"] is True
    assert by_label["plus5"]["n_lots"] < settings.sell_time_sensitivity_min_n_lots


# ── Test: bucket at exactly the threshold is NOT flagged (strict <) ─────────


def test_sell_time_sensitivity_bucket_at_threshold_not_flagged() -> None:
    """n_lots == threshold (10) is considered sufficient (strict less-than check)."""
    from app.core.config import settings

    bucket_rows = [
        _bucket_row("cheap", settings.sell_time_sensitivity_min_n_lots, 25),
        _bucket_row("median", 20, 30),
        _bucket_row("plus5", 20, 35),
        _bucket_row("plus10", 20, 40),
    ]
    db_mock = _make_db_mock(created_by="kopylov", bucket_rows=bucket_rows)
    app = _make_app()
    client = _client_with(app, db_mock, role="admin")
    data = _get(client)

    by_label = {b["price_premium_label"]: b for b in data["buckets"]}
    assert by_label["cheap"]["n_lots"] == settings.sell_time_sensitivity_min_n_lots
    assert by_label["cheap"]["insufficient_data"] is False


# ── Test: missing bucket (no DB rows at all) defaults n_lots=0 → flagged ────


def test_sell_time_sensitivity_missing_bucket_flagged() -> None:
    """A price bucket absent from the DB result (no matching lots) still comes
    back with n_lots=0 and MUST be flagged insufficient_data — never a bare 0
    presented as if it were a real, confident median."""
    bucket_rows = [
        _bucket_row("cheap", 20, 30),
        _bucket_row("median", 18, 35),
        _bucket_row("plus10", 15, 40),
        # 'plus5' entirely missing from bucket_rows.
    ]
    db_mock = _make_db_mock(created_by="kopylov", bucket_rows=bucket_rows)
    app = _make_app()
    client = _client_with(app, db_mock, role="admin")
    data = _get(client)

    by_label = {b["price_premium_label"]: b for b in data["buckets"]}
    assert by_label["plus5"]["n_lots"] == 0
    assert by_label["plus5"]["median_exposure_days"] is None
    assert by_label["plus5"]["insufficient_data"] is True


# ── Test: non-monotonic result is surfaced as-is, not smoothed ──────────────


def test_sell_time_sensitivity_does_not_smooth_non_monotonic_result() -> None:
    """#1995: +10% selling (median_exposure_days=20) FASTER than +5%
    (median_exposure_days=60) is a real, honestly-reported small-sample
    artifact — the endpoint must NOT invent smoothing/interpolation. Both
    thin buckets are simply flagged insufficient_data so the frontend can
    choose to de-emphasize them, and the raw (non-monotonic) numbers pass
    through unchanged."""
    bucket_rows = [
        _bucket_row("cheap", 20, 30),
        _bucket_row("median", 18, 35),
        _bucket_row("plus5", 4, 60),  # thin — slower
        _bucket_row("plus10", 3, 20),  # thin — faster (non-monotonic vs plus5)
    ]
    db_mock = _make_db_mock(created_by="kopylov", bucket_rows=bucket_rows)
    app = _make_app()
    client = _client_with(app, db_mock, role="admin")
    data = _get(client)

    by_label = {b["price_premium_label"]: b for b in data["buckets"]}
    # Raw numbers unchanged — no smoothing/reordering applied.
    assert by_label["plus5"]["median_exposure_days"] == 60
    assert by_label["plus10"]["median_exposure_days"] == 20
    # Both thin buckets honestly flagged instead of silently shown.
    assert by_label["plus5"]["insufficient_data"] is True
    assert by_label["plus10"]["insufficient_data"] is True


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(pytest.main([__file__, "-q"]))