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

"""Tests for GET /estimate/{id} and /estimate/{id}/pdf ownership scoping (#690 IDOR).

Closes cross-pilot data-leak: any authenticated pilot could read/download another
pilot's estimate (leaking client_name/client_phone). The endpoints now scope to the
owner (trade_in_estimates.created_by) OR admin, mirroring /history (#656).

Реальная БД не нужна: DB + get_role мокируются. Для 200-кейсов подменяем
_qc_geo_to_precision и generate_trade_in_pdf, чтобы не тащить тяжёлый estimator/PDF.
"""

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 = "11111111-1111-1111-1111-111111111111"


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

    original = auth_mod.get_role
    yield
    auth_mod.get_role = original


@pytest.fixture()
def trade_in_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 _make_estimate_row(created_by: str | None, retain_until: object = None) -> SimpleNamespace:
    """A trade_in_estimates row with the full column set the endpoints read.

    retain_until defaults to None (PR-D1, migration 240) -- unpaid, matches every
    row that existed before that migration; explicit param lets retention-gate
    tests (see test_estimate_retention_gate.py) construct a paid row.

    relaxations/reliability (migration 255) default to the schema defaults
    ('[]' / 'ok') -- matches every pre-migration row (no backfill). Revival
    scenarios (a "dead" median_price<=0/NULL row) are covered separately in
    test_estimate_revival.py with their own dedicated row builders, since this
    fixture's downstream tests here all assume a "live" estimate.
    """
    from datetime import UTC, datetime, timedelta

    return SimpleNamespace(
        id=_ESTIMATE_ID,
        median_price=5_000_000,
        range_low=4_500_000,
        range_high=5_500_000,
        median_price_per_m2=100_000,
        confidence="medium",
        confidence_explanation="ok",
        n_analogs=7,
        # #2899: колонка есть у всех строк после миграции 267; NULL у старых
        # (бэкфилла нет — позицию по сохранённому top-10 не восстановить).
        market_percentile=63,
        analogs=[],
        actual_deals=[],
        sources_used=["avito"],
        data_freshness_minutes=10,
        expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
        retain_until=retain_until,
        address="ул. Тестовая, 1",
        lat=56.8,
        lon=60.6,
        area_m2=50.0,
        rooms=2,
        floor=3,
        total_floors=9,
        year_built=2010,
        house_type="монолит",
        repair_state="хороший",
        has_balcony=True,
        canonical_address="ул. Тестовая, 1",
        house_cadnum=None,
        house_fias_id=None,
        dadata_qc_geo=0,
        dadata_metro=[],
        expected_sold_price=4_600_000,
        expected_sold_range_low=4_200_000,
        expected_sold_range_high=5_000_000,
        expected_sold_per_m2=92_000,
        asking_to_sold_ratio=0.85,
        ratio_basis="per_rooms",
        created_by=created_by,
        created_at=datetime.now(tz=UTC),
        relaxations=[],
        reliability="ok",
    )


def _make_db_mock(row: SimpleNamespace | None) -> MagicMock:
    """DB session mock returning *row* from .execute(...).fetchone()."""
    db = MagicMock()
    execute_result = MagicMock()
    execute_result.fetchone.return_value = row
    db.execute.return_value = execute_result
    return db


def _client_with(app: FastAPI, db_mock: MagicMock, role: str | None) -> TestClient:
    """Override get_db with *db_mock*; patch get_role to return *role* (or raise KeyError)."""
    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"]
    if role is None:

        def _raise_keyerror(_u: str):
            raise KeyError(_u)

        auth_mod.get_role = _raise_keyerror  # type: ignore[assignment]
    else:
        auth_mod.get_role = lambda _u: role  # type: ignore[assignment]
    return TestClient(app)


@pytest.fixture(autouse=True)
def _stub_precision_and_pdf():
    """Stub _qc_geo_to_precision + generate_trade_in_pdf so 200-paths don't pull heavy deps."""
    from app.api.v1 import trade_in as trade_in_module

    # #696: GET-rehydrate pulls four POST-only helpers from estimator. Stub them to
    # None by default (graceful "no data" path); the positive rehydrate test overrides
    # them per-call with data-returning lambdas.
    estimator_stub = SimpleNamespace(
        _qc_geo_to_precision=lambda _qc: "house",
        _fetch_price_trend=lambda *a, **k: None,
        _fetch_dkp_corridor=lambda *a, **k: None,
        _fetch_house_imv_anchor=lambda *a, **k: None,
        # (oblast C2): GET-rehydrate also resolves the target city (for
        # _fetch_dkp_corridor city-scoping) before calling the corridor helper.
        _resolve_target_city=lambda *a, **k: None,
        # #2043 (BE-1): GET-rehydrate also recomputes cv / source_counts from the
        # persisted analogs. Empty analogs in the fixture → None / {} (real behaviour).
        _cv_from_ppm2=lambda *a, **k: None,
        _source_counts=lambda *a, **k: {},
        # #2087 (M1): GET-rehydrate derives canonical sources_used via the shared
        # helper. Empty analogs + no valuation flags → [] (real behaviour).
        _canonical_sources=lambda *a, **k: [],
        # #2632: GET-rehydrate реконструирует фактический радиус подбора. У этих
        # фикстур нет ни подписи каскада, ни расстояний → None и есть настоящее
        # поведение (см. tests/test_estimator_search_radius_2632.py).
        rehydrate_search_radius_m=lambda *a, **k: None,
    )
    real_estimator = sys.modules.get("app.services.estimator")
    sys.modules["app.services.estimator"] = estimator_stub  # type: ignore[assignment]

    original_pdf = trade_in_module.generate_trade_in_pdf
    trade_in_module.generate_trade_in_pdf = lambda *a, **k: b"%PDF-fake"  # type: ignore[assignment]

    # brand resolution also imports lazily — stub the module it imports from.
    brand_stub = SimpleNamespace(get_brand=lambda _b, _db: SimpleNamespace(slug="generic"))
    real_brand = sys.modules.get("app.services.brand")
    sys.modules["app.services.brand"] = brand_stub  # type: ignore[assignment]

    yield

    trade_in_module.generate_trade_in_pdf = original_pdf
    if real_estimator is not None:
        sys.modules["app.services.estimator"] = real_estimator
    else:
        sys.modules.pop("app.services.estimator", None)
    if real_brand is not None:
        sys.modules["app.services.brand"] = real_brand
    else:
        sys.modules.pop("app.services.brand", None)


# ── GET /estimate/{id} ───────────────────────────────────────────────────────


def test_get_estimate_owner_can_read(trade_in_app: FastAPI) -> None:
    """Owner pilot reads own estimate → 200."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json()["estimate_id"] == _ESTIMATE_ID


def test_get_estimate_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Non-owner pilot must NOT read someone else's estimate → 404 (hide existence)."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


def test_get_estimate_admin_can_read_any(trade_in_app: FastAPI) -> None:
    """Admin reads any estimate regardless of owner → 200."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="someone_else"))
    client = _client_with(trade_in_app, db_mock, role="admin")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200


def test_get_estimate_rehydrates_post_only_fields(trade_in_app: FastAPI) -> None:
    """#696: GET /{id} recomputes price_trend / avito_imv / dkp_corridor / last_scraped_at.

    Previously these POST-only derived fields came back null on shared-link / PDF /
    ?id= restore. Stub the estimator helpers with data and assert they surface.
    """
    est = sys.modules["app.services.estimator"]
    est._fetch_price_trend = lambda *a, **k: [  # type: ignore[attr-defined]
        {"month": "2026-03", "ppm2": 140_000},
        {"month": "2026-04", "ppm2": 150_000},
        {"month": "2026-05", "ppm2": 160_000},
    ]
    est._fetch_house_imv_anchor = lambda *a, **k: {  # type: ignore[attr-defined]
        "recommended_price": 5_200_000,
        "lower_price": 4_800_000,
        "higher_price": 5_600_000,
        "market_count": 42,
    }
    est._fetch_dkp_corridor = lambda *a, **k: {  # type: ignore[attr-defined]
        "count": 5,
        "low_ppm2": 120_000,
        "median_ppm2": 150_000,
        "high_ppm2": 180_000,
        "period_months": 12,
    }

    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert len(body["price_trend"]) == 3
    assert body["avito_imv"]["recommended_price"] == 5_200_000
    assert body["dkp_corridor"]["count"] == 5
    # last_scraped_at = created_at − data_freshness_minutes (both persisted) → non-null
    assert body["last_scraped_at"] is not None


def test_get_estimate_requires_authenticated_user(trade_in_app: FastAPI) -> None:
    """No X-Authenticated-User header → 401."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}")
    assert resp.status_code == 401


def test_get_estimate_unknown_user_gets_403(trade_in_app: FastAPI) -> None:
    """Authenticated via Caddy but missing from roles.yaml → 403."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role=None)
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "ghost"},
    )
    assert resp.status_code == 403


# ── GET /estimate/{id}/pdf ───────────────────────────────────────────────────


def test_pdf_owner_can_download(trade_in_app: FastAPI) -> None:
    """Owner pilot downloads own PDF → 200."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.headers["content-type"] == "application/pdf"


def test_pdf_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Non-owner pilot must NOT download someone else's PDF → 404."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


def test_pdf_admin_can_download_any(trade_in_app: FastAPI) -> None:
    """Admin downloads any PDF regardless of owner → 200."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="someone_else"))
    client = _client_with(trade_in_app, db_mock, role="admin")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200


def test_pdf_requires_authenticated_user(trade_in_app: FastAPI) -> None:
    """No X-Authenticated-User header → 401."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf")
    assert resp.status_code == 401


def test_pdf_unknown_user_gets_403(trade_in_app: FastAPI) -> None:
    """/pdf: authenticated via Caddy but missing from roles.yaml → 403 (reviewer gap)."""
    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role=None)
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "ghost"},
    )
    assert resp.status_code == 403


# ── Reviewer-named gap: legacy NULL created_by ───────────────────────────────


def test_get_estimate_legacy_null_owner_non_admin_gets_404(trade_in_app: FastAPI) -> None:
    """Legacy estimate with NULL created_by: non-admin pilot must NOT read it → 404."""
    db_mock = _make_db_mock(_make_estimate_row(created_by=None))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 404


# ── Photo endpoints (#690 — raw apartment-interior bytes are enumerable PII) ──


def _make_photo_row() -> SimpleNamespace:
    """A minimal estimate_photos row for get_photo (content + content_type)."""
    return SimpleNamespace(content=b"\x89PNG-fake", content_type="image/png")


def _make_db_mock_seq(*rows: SimpleNamespace | None) -> MagicMock:
    """DB session mock whose successive .execute(...).fetchone() yield *rows* in order.

    Derived/photo routes run the guard SELECT (created_by) first, then their own
    queries; configure side_effect so the guard row comes first.
    """
    db = MagicMock()

    def _execute(*_a, **_k):
        result = MagicMock()
        result.fetchone.return_value = next(_iter)
        return result

    _iter = iter(rows)
    db.execute.side_effect = _execute
    return db


def test_get_photo_owner_can_read(trade_in_app: FastAPI) -> None:
    """Owner pilot reads photo bytes → 200."""
    db_mock = _make_db_mock_seq(
        SimpleNamespace(created_by="kopylov"),  # guard query
        _make_photo_row(),  # route query
    )
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.content == b"\x89PNG-fake"


def test_get_photo_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Non-owner pilot must NOT read someone else's photo → 404 (guard raises first)."""
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


def test_get_photo_admin_can_read_any(trade_in_app: FastAPI) -> None:
    """Admin reads any photo → 200."""
    db_mock = _make_db_mock_seq(
        SimpleNamespace(created_by="someone_else"),
        _make_photo_row(),
    )
    client = _client_with(trade_in_app, db_mock, role="admin")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200


def test_get_photo_requires_authenticated_user(trade_in_app: FastAPI) -> None:
    """get_photo: no X-Authenticated-User header → 401."""
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="kopylov"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
    )
    assert resp.status_code == 401


def test_list_photos_owner_can_read(trade_in_app: FastAPI) -> None:
    """Owner pilot lists photos → 200 (empty list ok)."""
    db = MagicMock()
    guard_result = MagicMock()
    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
    list_result = MagicMock()
    list_result.mappings.return_value.all.return_value = []
    db.execute.side_effect = [guard_result, list_result]
    client = _client_with(trade_in_app, db, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json() == []


def test_list_photos_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Non-owner pilot must NOT list someone else's photos → 404."""
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


def test_upload_photo_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Write-side IDOR: non-owner pilot must NOT upload to someone else's estimate → 404.

    Guard runs on the existence SELECT (created_by) before any write — so a single
    fetchone returning created_by='admin' + a non-owner header is enough.
    """
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="admin"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.post(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
        headers={"X-Authenticated-User": "attacker"},
        files={"file": ("x.png", b"\x89PNG-fake", "image/png")},
    )
    assert resp.status_code == 404


def test_upload_photo_owner_can_upload(trade_in_app: FastAPI, monkeypatch) -> None:
    """Owner pilot uploads → 200. Stub sanitize_image + the INSERT mapping row."""
    from app.api.v1 import trade_in as trade_in_module

    monkeypatch.setattr(
        trade_in_module, "sanitize_image", lambda _c: (b"\x89PNG-clean", "image/png")
    )

    from datetime import UTC, datetime

    db = MagicMock()
    guard_result = MagicMock()
    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
    count_result = MagicMock()
    count_result.scalar_one.return_value = 0
    insert_result = MagicMock()
    insert_result.mappings.return_value.fetchone.return_value = {
        "id": _ESTIMATE_ID,
        "filename": "x.png",
        "content_type": "image/png",
        "size_bytes": 9,
        "uploaded_at": datetime.now(tz=UTC),
    }
    db.execute.side_effect = [guard_result, count_result, insert_result]
    client = _client_with(trade_in_app, db, role="pilot")
    resp = client.post(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
        headers={"X-Authenticated-User": "kopylov"},
        files={"file": ("x.png", b"\x89PNG-fake", "image/png")},
    )
    assert resp.status_code == 200


class _InfiniteUploadFile:
    """UploadFile stub whose async read(size) yields endless 64KB chunks (#2233).

    Simulates a multi-GB stream. Counts read() calls so the test can assert the
    handler stops reading right after the cap is exceeded instead of draining the
    whole body into RAM.
    """

    content_type = "image/png"
    filename = "huge.png"

    def __init__(self) -> None:
        self.read_calls = 0

    async def read(self, size: int = -1) -> bytes:
        self.read_calls += 1
        return b"\x00" * (64 * 1024)


async def test_upload_photo_streams_over_cap_returns_413_without_full_read() -> None:
    """#2233: a stream larger than 10 MB → 413 raised early, body NOT fully read.

    Acceptance: read() is called only ~(10MB/64KB)+1 times (161), proving the loop
    bails on the first chunk that pushes total past _MAX_PHOTO_BYTES rather than
    buffering an unbounded upload (which would OOM the 768m-capped container, #2214).
    """
    from uuid import UUID

    from fastapi import HTTPException

    import app.core.auth as auth_mod
    from app.api.v1.trade_in import _MAX_PHOTO_BYTES, upload_photo

    auth_mod.get_role = lambda _u: "pilot"  # type: ignore[assignment]

    # guard SELECT → owner; count SELECT → 0. INSERT must never be reached.
    db = MagicMock()
    guard_result = MagicMock()
    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
    count_result = MagicMock()
    count_result.scalar_one.return_value = 0
    db.execute.side_effect = [guard_result, count_result]

    upload = _InfiniteUploadFile()

    with pytest.raises(HTTPException) as exc_info:
        await upload_photo(
            estimate_id=UUID(_ESTIMATE_ID),
            db=db,
            file=upload,  # type: ignore[arg-type]
            x_authenticated_user="kopylov",
        )

    assert exc_info.value.status_code == 413
    expected_reads = _MAX_PHOTO_BYTES // (64 * 1024) + 1  # 161
    assert upload.read_calls == expected_reads
    # sanity: bounded, nowhere near infinite
    assert upload.read_calls < 200
    # INSERT (3rd execute) never fired — nothing persisted
    assert db.execute.call_count == 2
    db.commit.assert_not_called()


# ── Derived analytics routes (representative: /houses, /imv-benchmark) ────────


def test_get_estimate_houses_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """Derived route: non-owner pilot blocked by guard → 404 (before any house query)."""
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/houses",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


def test_get_estimate_houses_owner_can_read(trade_in_app: FastAPI) -> None:
    """Derived route: owner pilot passes guard → 200. Stub target with no address/geo."""
    db = MagicMock()
    guard_result = MagicMock()
    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
    target_result = MagicMock()
    target_result.fetchone.return_value = SimpleNamespace(lat=None, lon=None, address=None)
    db.execute.side_effect = [guard_result, target_result]
    client = _client_with(trade_in_app, db, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/houses",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json() == []


def test_get_estimate_imv_benchmark_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
    """imv-benchmark: non-owner pilot blocked by guard → 404."""
    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/imv-benchmark",
        headers={"X-Authenticated-User": "attacker"},
    )
    assert resp.status_code == 404


# ── Payments retention: retention gate unification (retain_until, PR #2754) ──


def test_estimate_readable_sql_uses_disjunction() -> None:
    """Single definition — OR retain_until, not a hand-copied expression."""
    from app.api.v1.trade_in import ESTIMATE_READABLE_SQL

    assert "expires_at > NOW()" in ESTIMATE_READABLE_SQL
    assert "retain_until > NOW()" in ESTIMATE_READABLE_SQL
    assert " OR " in ESTIMATE_READABLE_SQL


def test_get_estimate_sql_built_from_shared_constant() -> None:
    """GET /estimate/{id} SQL filter is built FROM ESTIMATE_READABLE_SQL, not a
    hand-copied literal — regression guard against the two gates drifting apart
    again (that's exactly what happened before this PR: 404 here, 410 in /pdf).

    Inspects `load_estimate`, not the `get_estimate` route: the payments PR moved
    the body there so the paid capability link (`/r/<token>`, app/api/v1/
    payments.py) reuses the SAME loader instead of growing a third copy of the
    readability gate — which is precisely what this guard exists to prevent.
    """
    import inspect

    from app.api.v1.trade_in import load_estimate

    src = inspect.getsource(load_estimate)
    assert "ESTIMATE_READABLE_SQL" in src
    assert "expires_at > NOW()" not in src, "hand-copied predicate, not the shared constant"
    assert "retain_until" in src, "SELECT must also fetch retain_until"


def test_estimate_pdf_select_includes_retain_until_column() -> None:
    import inspect

    from app.api.v1.trade_in import estimate_pdf

    assert "retain_until" in inspect.getsource(estimate_pdf)


@pytest.mark.parametrize(
    ("expires_delta_hours", "retain_delta_days", "expected"),
    [
        (12, None, True),  # not expired, unpaid — current B2B/B2C behaviour, unchanged
        (-1, None, False),  # expired, unpaid — current behaviour (404/410), unchanged
        (-1, 365, True),  # expired but PAID — new: readable
        (12, 365, True),  # not expired AND paid — readable
        (-1, -1, False),  # expired, and the (hypothetical) retain_until also in the past
    ],
)
def test_estimate_readable_truth_table(
    expires_delta_hours: int, retain_delta_days: int | None, expected: bool
) -> None:
    from datetime import UTC, datetime, timedelta

    from app.api.v1.trade_in import estimate_readable

    expires_at = datetime.now(tz=UTC) + timedelta(hours=expires_delta_hours)
    retain_until = (
        datetime.now(tz=UTC) + timedelta(days=retain_delta_days)
        if retain_delta_days is not None
        else None
    )
    assert estimate_readable(expires_at, retain_until) is expected


def test_pdf_expired_but_paid_returns_200(trade_in_app: FastAPI) -> None:
    """expires_at in the past, retain_until in the future → PDF still downloads
    (200). Exactly the scenario PR-D1 exists for: a paid report must outlive
    the 24h expires_at link TTL."""
    from datetime import UTC, datetime, timedelta

    row = _make_estimate_row(created_by="kopylov")
    row.expires_at = datetime.now(tz=UTC) - timedelta(hours=1)
    row.retain_until = datetime.now(tz=UTC) + timedelta(days=300)
    db_mock = _make_db_mock(row)
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.headers["content-type"] == "application/pdf"


def test_pdf_expired_unpaid_returns_410_without_ttl_text(trade_in_app: FastAPI) -> None:
    """expires_at in the past, retain_until NULL (unpaid, unchanged behaviour) →
    410, and the detail text no longer claims a specific '24h TTL' (would be a
    lie now that retain_until exists for paid rows)."""
    from datetime import UTC, datetime, timedelta

    row = _make_estimate_row(created_by="kopylov")
    row.expires_at = datetime.now(tz=UTC) - timedelta(hours=1)
    row.retain_until = None
    db_mock = _make_db_mock(row)
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 410
    assert resp.json()["detail"] == "estimate expired"
    assert "24h" not in resp.json()["detail"]
    assert "TTL" not in resp.json()["detail"]


def test_get_estimate_response_includes_retain_until_field(trade_in_app: FastAPI) -> None:
    """Response schema exposes retain_until (nullable) — schemas/trade_in.py."""
    row = _make_estimate_row(created_by="kopylov")  # retain_until defaults to None
    db_mock = _make_db_mock(row)
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json()["retain_until"] is None


def test_get_estimate_surfaces_market_percentile(trade_in_app: FastAPI) -> None:
    """#2899: позиция в когорте переживает перезагрузку по ссылке и уходит в PDF.

    Значение считается только на POST (когорты в БД нет — в `analogs` лежит top-10,
    а не выборка), поэтому оно ОБЯЗАНО храниться в колонке и подниматься обоими SELECT'ами.
    У GET и PDF списки колонок РАЗНЫЕ и живут в разных функциях — один общий тест их
    не покрывает, отсюда две проверки.
    """
    row = _make_estimate_row(created_by="kopylov")
    db_mock = _make_db_mock(row)
    client = _client_with(trade_in_app, db_mock, role="pilot")

    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json()["market_percentile"] == 63


def test_get_estimate_market_percentile_nullable(trade_in_app: FastAPI) -> None:
    """Контроль: NULL проходит как null, а не роняет ответ.

    Так выглядят все строки до миграции 267 и все оценки с когортой меньше 15 лотов.
    Зелёный с обеих сторон правки — доказывает, что поле необязательное.
    """
    row = _make_estimate_row(created_by="kopylov")
    row.market_percentile = None
    db_mock = _make_db_mock(row)
    client = _client_with(trade_in_app, db_mock, role="pilot")

    resp = client.get(
        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json()["market_percentile"] is None