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

"""Regression tests for the repair-state coefficient (issue #7).

Verifies the `_REPAIR_COEF` heuristic multiplier in `estimate_quality`:
- excellent / needs_repair median ratio == coef ratio (within int rounding)
- standard (baseline 1.0) is a true no-op vs the unadjusted market median
- repair_state=None is a no-op (coef 1.0)
- coef != 1.0 emits the repair note into confidence_explanation

Coefficient values are read DYNAMICALLY from `_REPAIR_COEF` so the assertions
survive future re-calibration. The full `estimate_quality` flow is exercised with
all I/O stubbed (geocode / house-meta / _fetch_analogs / _fetch_deals / IMV /
Yandex / Cian / DB) so the test isolates the multiplier — no DB, no network.
"""

from __future__ import annotations

import os
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import anyio

# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")


# ── Fixtures / helpers ──────────────────────────────────────────────────────


def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, Any]:
    """Minimal listing dict matching the keys the aggregation block reads."""
    price_rub = price_per_m2 * area_m2
    return {
        "source": "cian",
        "source_url": "https://cian.ru/offer/1",
        "address": "ЕКБ, ул. Учителей, 18",
        "lat": 56.838,
        "lon": 60.595,
        "rooms": 1,
        "area_m2": area_m2,
        "floor": 4,
        "total_floors": 16,
        "price_rub": price_rub,
        "price_per_m2": price_per_m2,
        "listing_date": datetime(2026, 5, 1),
        "days_on_market": 10,
        "photo_urls": [],
        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
        "distance_m": 100.0,
        "relevance_score": 0.1,
    }


# #oblast-E: 5 fixed analogs (>= HEADLINE_LISTINGS_MIN_N so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic runs) → deterministic median ppm2 = 150_000 (symmetric around it,
# < outlier-drop threshold).
_ANALOGS: list[dict[str, Any]] = [
    _make_listing(price_per_m2=135_000.0),
    _make_listing(price_per_m2=145_000.0),
    _make_listing(price_per_m2=150_000.0),
    _make_listing(price_per_m2=155_000.0),
    _make_listing(price_per_m2=165_000.0),
]


def _make_fake_geo():
    from app.services.geocoder import GeocodeResult

    return GeocodeResult(
        lat=56.838,
        lon=60.595,
        full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
        provider="nominatim",
    )


def _make_payload(repair_state: str | None):
    from app.schemas.trade_in import TradeInEstimateInput

    return TradeInEstimateInput(
        address="ЕКБ, ул. Учителей, 18",
        area_m2=40.0,
        rooms=1,
        floor=4,
        total_floors=16,
        repair_state=repair_state,
    )


def _run_estimate(repair_state: str | None):
    """Invoke estimate_quality with every external dependency stubbed."""
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload(repair_state)

    async def _run():
        with (
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
            patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
            patch("app.services.estimator.match_house_readonly", return_value=None),
            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
            # 3-tuple: (listings, fallback_used, analog_tier). Same analogs every call so
            # all fallback tiers are equivalent and the median is stable.
            patch(
                "app.services.estimator._fetch_analogs", return_value=(list(_ANALOGS), False, "S")
            ),
            patch("app.services.estimator._fetch_deals", return_value=[]),
            patch(
                "app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)
            ),
            patch(
                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
                new=AsyncMock(return_value=None),
            ),
            patch(
                "app.services.estimator.estimate_via_cian_valuation",
                new=AsyncMock(return_value=None),
            ),
            # #648 S3: stub asking→sold lookup off so this test isolates the
            # repair coefficient (no sold-correction, no DB).
            patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
        ):
            return await estimate_quality(payload, db)

    return anyio.run(_run)


# ── Tests ───────────────────────────────────────────────────────────────────


def test_excellent_to_needs_repair_ratio_matches_coef() -> None:
    """excellent ÷ needs_repair median ≈ _REPAIR_COEF["excellent"] / [...]["needs_repair"]."""
    from app.services.estimator import _REPAIR_COEF

    excellent = _run_estimate("excellent")
    needs_repair = _run_estimate("needs_repair")

    expected_ratio = _REPAIR_COEF["excellent"] / _REPAIR_COEF["needs_repair"]
    actual_ratio = excellent.median_price_rub / needs_repair.median_price_rub

    # int() truncation on both medians ⇒ allow a small tolerance.
    assert abs(actual_ratio - expected_ratio) < 0.005, (
        f"excellent/needs_repair ratio {actual_ratio:.5f} != coef ratio {expected_ratio:.5f}"
    )


def test_standard_is_baseline_noop() -> None:
    """repair_state='standard' → median equals the unadjusted market median (coef 1.0)."""
    from app.services.estimator import _REPAIR_COEF

    assert _REPAIR_COEF["standard"] == 1.0, "baseline invariant: standard must be 1.0"

    standard = _run_estimate("standard")
    baseline = _run_estimate(None)

    # Unadjusted median: median ppm2 (150_000) × area (40.0) = 6_000_000.
    expected_median = int(150_000.0 * 40.0)
    assert standard.median_price_rub == expected_median
    assert standard.median_price_rub == baseline.median_price_rub


def test_none_repair_state_is_noop() -> None:
    """repair_state=None → coef 1.0, no adjustment, no repair note."""
    baseline = _run_estimate(None)

    expected_median = int(150_000.0 * 40.0)
    assert baseline.median_price_rub == expected_median
    assert "скорректирована на состояние ремонта" not in (baseline.confidence_explanation or "")


def test_repair_note_present_when_coef_differs() -> None:
    """coef != 1.0 (excellent) → confidence_explanation contains the repair note."""
    from app.services.estimator import _REPAIR_COEF

    assert _REPAIR_COEF["excellent"] != 1.0, "fixture invariant: excellent must adjust"

    excellent = _run_estimate("excellent")
    explanation = excellent.confidence_explanation or ""
    assert "скорректирована на состояние ремонта" in explanation
    # Note reports the signed percentage derived from the coefficient.
    pct = round((_REPAIR_COEF["excellent"] - 1.0) * 100)
    assert f"{pct:+d}%" in explanation


def test_standard_emits_no_repair_note() -> None:
    """Baseline standard (1.0) is a true no-op — no repair note appended."""
    standard = _run_estimate("standard")
    assert "скорректирована на состояние ремонта" not in (standard.confidence_explanation or "")