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) -> SimpleNamespace:
    """A trade_in_estimates row with the full column set the endpoints read."""
    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,
        analogs=[],
        actual_deals=[],
        sources_used=["avito"],
        data_freshness_minutes=10,
        expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
        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,
    )


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

    estimator_stub = SimpleNamespace(_qc_geo_to_precision=lambda _qc: "house")
    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_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