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


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


# ── 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