diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py index 42b75cd3..88b8387f 100644 --- a/tradein-mvp/backend/app/api/v1/trade_in.py +++ b/tradein-mvp/backend/app/api/v1/trade_in.py @@ -44,6 +44,36 @@ logger = logging.getLogger(__name__) router = APIRouter() +def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None: + """IDOR guard (#690): только владелец оценки или admin могут её читать. + + Зеркало скоупинга /history (#656). 401 — нет заголовка X-Authenticated-User + (Caddy basic_auth обязателен); 403 — юзер отсутствует в roles.yaml; 404 — + pilot читает чужую оценку (скрываем существование, не подтверждаем чужой id). + """ + if not x_authenticated_user: + raise HTTPException( + status_code=401, + detail="no authenticated user (Caddy basic_auth required)", + ) + + from app.core.auth import get_role + + try: + role = get_role(x_authenticated_user) + except KeyError: + logger.warning( + "user %r authenticated via Caddy but missing from roles.yaml", + x_authenticated_user, + ) + raise HTTPException(status_code=403, detail="user not in roles config") from None + + if role == "admin": + return + if created_by is None or created_by != x_authenticated_user: + raise HTTPException(status_code=404, detail="estimate not found") + + @router.post("/estimate", response_model=AggregatedEstimate) async def estimate( payload: TradeInEstimateInput, @@ -100,6 +130,7 @@ def get_quota( def get_estimate( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], + x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, ) -> AggregatedEstimate: """Получить сохранённую оценку по UUID (для генерации PDF). @@ -118,7 +149,7 @@ def get_estimate( dadata_qc_geo, dadata_metro, expected_sold_price, expected_sold_range_low, expected_sold_range_high, expected_sold_per_m2, - asking_to_sold_ratio, ratio_basis + asking_to_sold_ratio, ratio_basis, created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid) AND expires_at > NOW() @@ -130,6 +161,8 @@ def get_estimate( if row is None: raise HTTPException(status_code=404, detail="estimate not found or expired") + _assert_estimate_access(row.created_by, x_authenticated_user) + from app.services.estimator import _qc_geo_to_precision analogs = [AnalogLot(**a) for a in (row.analogs or [])] @@ -190,6 +223,7 @@ def estimate_pdf( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], brand: str | None = None, + x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, ) -> Response: """Скачать 4-страничный PDF-отчёт для оценки trade-in. @@ -210,7 +244,7 @@ def estimate_pdf( dadata_qc_geo, dadata_metro, expected_sold_price, expected_sold_range_low, expected_sold_range_high, expected_sold_per_m2, - asking_to_sold_ratio, ratio_basis + asking_to_sold_ratio, ratio_basis, created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ @@ -221,6 +255,8 @@ def estimate_pdf( if row is None: raise HTTPException(status_code=404, detail="estimate not found") + _assert_estimate_access(row.created_by, x_authenticated_user) + if row.expires_at.replace(tzinfo=UTC) < datetime.now(tz=UTC): raise HTTPException(status_code=410, detail="estimate expired (24h TTL)") @@ -275,12 +311,15 @@ def estimate_pdf( } from app.services.brand import get_brand as _resolve_brand + brand_obj = _resolve_brand(brand, db) pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj) filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf" logger.info( "PDF generated estimate_id=%s brand=%s size=%d", - estimate_id, brand_obj.slug, len(pdf_bytes), + estimate_id, + brand_obj.slug, + len(pdf_bytes), ) return Response( content=pdf_bytes, @@ -290,7 +329,7 @@ def estimate_pdf( # ── Фото квартиры (#394) ───────────────────────────────────────────────────── -_MAX_PHOTO_BYTES = 10 * 1024 * 1024 # 10 МБ на фото +_MAX_PHOTO_BYTES = 10 * 1024 * 1024 # 10 МБ на фото _MAX_PHOTOS_PER_ESTIMATE = 12 _ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/heic"} @@ -337,27 +376,35 @@ async def upload_photo( except ImageSanitizationError as e: raise HTTPException(status_code=400, detail=str(e)) from e - row = db.execute( - text( - """ + row = ( + db.execute( + text( + """ INSERT INTO estimate_photos (estimate_id, filename, content_type, content, size_bytes) VALUES (CAST(:eid AS uuid), :fn, :ct, :content, :sz) RETURNING id, filename, content_type, size_bytes, uploaded_at """ - ), - { - "eid": str(estimate_id), - "fn": file.filename, - "ct": sanitized_ctype, - "content": sanitized_bytes, - "sz": len(sanitized_bytes), - }, - ).mappings().fetchone() + ), + { + "eid": str(estimate_id), + "fn": file.filename, + "ct": sanitized_ctype, + "content": sanitized_bytes, + "sz": len(sanitized_bytes), + }, + ) + .mappings() + .fetchone() + ) db.commit() logger.info( "photo uploaded: estimate=%s photo=%s orig_size=%d sanitized_size=%d ctype=%s", - estimate_id, row["id"], len(content), len(sanitized_bytes), sanitized_ctype, + estimate_id, + row["id"], + len(content), + len(sanitized_bytes), + sanitized_ctype, ) return PhotoMeta(**row) @@ -368,17 +415,21 @@ def list_photos( db: Annotated[Session, Depends(get_db)], ) -> list[PhotoMeta]: """Список фото оценки — метаданные, без содержимого (#394).""" - rows = db.execute( - text( - """ + rows = ( + db.execute( + text( + """ SELECT id, filename, content_type, size_bytes, uploaded_at FROM estimate_photos WHERE estimate_id = CAST(:id AS uuid) ORDER BY uploaded_at """ - ), - {"id": str(estimate_id)}, - ).mappings().all() + ), + {"id": str(estimate_id)}, + ) + .mappings() + .all() + ) return [PhotoMeta(**r) for r in rows] @@ -452,9 +503,10 @@ def estimate_history( where = "WHERE created_by = :owner" params["owner"] = x_authenticated_user - rows = db.execute( - text( - f""" + rows = ( + db.execute( + text( + f""" SELECT id, address, rooms, area_m2, median_price, confidence, n_analogs, created_at FROM trade_in_estimates @@ -462,18 +514,22 @@ def estimate_history( ORDER BY created_at DESC LIMIT :limit """ - ), - params, - ).mappings().all() + ), + params, + ) + .mappings() + .all() + ) return [dict(r) for r in rows] @router.get("/cache-stats") def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]: """Состояние данных и кэшей (#399) — для страницы «Кэш».""" - row = db.execute( - text( - """ + row = ( + db.execute( + text( + """ SELECT (SELECT count(*) FROM geocode_cache) AS geocode_cache, (SELECT count(*) FROM geocode_cache WHERE expires_at > NOW()) @@ -485,8 +541,11 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]: (SELECT count(*) FROM house_metadata) AS house_metadata, (SELECT count(*) FROM trade_in_estimates) AS estimates_total """ + ) ) - ).mappings().fetchone() + .mappings() + .fetchone() + ) return dict(row) if row else {} @@ -546,7 +605,9 @@ def get_estimate_houses( """ ), {"addr": target.address}, - ).mappings().all() + ) + .mappings() + .all() ) # Path 2: geo-nearby (any source, 500м radius) @@ -573,7 +634,9 @@ def get_estimate_houses( """ ), {"lat": target.lat, "lon": target.lon}, - ).mappings().all() + ) + .mappings() + .all() ) # Merge: direct первым, затем nearby (dedup by house_id) @@ -645,9 +708,10 @@ def get_estimate_placement_history( if not house_ids: return [] - history = db.execute( - text( - """ + history = ( + db.execute( + text( + """ SELECT id, source, house_id, ext_item_id, title, rooms, area_m2, floor, total_floors, start_price, start_price_date, last_price, last_price_date, removed_date, exposure_days, @@ -657,9 +721,12 @@ def get_estimate_placement_history( ORDER BY COALESCE(last_price_date, start_price_date) DESC NULLS LAST LIMIT 50 """ - ), - {"house_ids": house_ids}, - ).mappings().all() + ), + {"house_ids": house_ids}, + ) + .mappings() + .all() + ) return [PlacementHistoryEntry(**dict(r)) for r in history] @@ -740,9 +807,10 @@ def get_estimate_house_analytics( ) # 3. Price history by year × source (median ₽/м²) - price_history_rows = db.execute( - text( - """ + price_history_rows = ( + db.execute( + text( + """ SELECT EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year, source, @@ -758,14 +826,18 @@ def get_estimate_house_analytics( GROUP BY year, source ORDER BY year ASC, source ASC """ - ), - {"ids": house_ids}, - ).mappings().all() + ), + {"ids": house_ids}, + ) + .mappings() + .all() + ) # 4. Recent sold (12 months, with removed_date) - recent_sold_rows = db.execute( - text( - """ + recent_sold_rows = ( + db.execute( + text( + """ SELECT id, source, rooms, area_m2, floor, start_price, last_price, removed_date, exposure_days, CASE WHEN start_price > 0 AND last_price IS NOT NULL @@ -778,14 +850,18 @@ def get_estimate_house_analytics( ORDER BY removed_date DESC LIMIT 20 """ - ), - {"ids": house_ids}, - ).mappings().all() + ), + {"ids": house_ids}, + ) + .mappings() + .all() + ) # 5. KPI aggregate - kpi_row = db.execute( - text( - """ + kpi_row = ( + db.execute( + text( + """ SELECT COUNT(*) AS total_lots, COUNT(*) FILTER (WHERE removed_date IS NOT NULL) AS sold_count, @@ -811,9 +887,12 @@ def get_estimate_house_analytics( FROM house_placement_history WHERE house_id = ANY(:ids) """ - ), - {"ids": house_ids}, - ).mappings().first() + ), + {"ids": house_ids}, + ) + .mappings() + .first() + ) return HouseAnalyticsResponse( house_ids=house_ids, @@ -857,9 +936,10 @@ def get_estimate_cian_price_changes( Возвращает только аналоги с хотя бы одним изменением цены. Пустой список если нет cian-аналогов или нет истории. """ - rows = db.execute( - text( - """ + rows = ( + db.execute( + text( + """ WITH cian_analogs AS ( SELECT DISTINCT substring(a->>'source_url' from '/sale/flat/(\\d+)/') AS cian_id @@ -916,9 +996,12 @@ def get_estimate_cian_price_changes( WHERE ca.n_changes > 0 ORDER BY ca.last_change_time DESC """ - ), - {"eid": str(estimate_id)}, - ).mappings().all() + ), + {"eid": str(estimate_id)}, + ) + .mappings() + .all() + ) return [CianPriceChangeStats(**dict(r)) for r in rows] @@ -1014,9 +1097,10 @@ def get_estimate_sell_time_sensitivity( ).scalar() # 3. Per-year median (для расчёта premium per lot); используем CTE для bucket-расчёта - bucket_rows = db.execute( - text( - """ + bucket_rows = ( + db.execute( + text( + """ WITH year_medians AS ( SELECT EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year, @@ -1070,9 +1154,12 @@ def get_estimate_sell_time_sensitivity( WHERE bucket IS NOT NULL GROUP BY bucket """ - ), - {"ids": house_ids}, - ).mappings().all() + ), + {"ids": house_ids}, + ) + .mappings() + .all() + ) # 4. Build buckets — гарантируем все 4 даже если данных нет в bucket bucket_map = {r["bucket"]: dict(r) for r in bucket_rows} @@ -1238,9 +1325,10 @@ def get_street_deals( area_min = area_m2 * (1.0 - area_tolerance) area_max = area_m2 * (1.0 + area_tolerance) - rows = db.execute( - text( - """ + rows = ( + db.execute( + text( + """ SELECT address, area_m2, rooms, floor, total_floors, price_rub, price_per_m2, deal_date, source FROM deals @@ -1252,20 +1340,26 @@ def get_street_deals( AND price_rub > 0 ORDER BY deal_date DESC """ - ), - { - "street_pattern": "%" + street_name + "%", - "rooms": rooms, - "area_min": area_min, - "area_max": area_max, - "period_months": period_months, - }, - ).mappings().all() + ), + { + "street_pattern": "%" + street_name + "%", + "rooms": rooms, + "area_min": area_min, + "area_max": area_max, + "period_months": period_months, + }, + ) + .mappings() + .all() + ) if not rows: logger.info( "street-deals: no rows found street=%r rooms=%d area=%.1f±%.0f%%", - street_name, rooms, area_m2, area_tolerance * 100, + street_name, + rooms, + area_m2, + area_tolerance * 100, ) return StreetDealsResponse( street=street_name, @@ -1294,7 +1388,11 @@ def get_street_deals( logger.info( "street-deals: street=%r rooms=%d area=%.1f count=%d median_ppm2=%.0f", - street_name, rooms, area_m2, count, median_ppm2, + street_name, + rooms, + area_m2, + count, + median_ppm2, ) return StreetDealsResponse( @@ -1357,9 +1455,10 @@ def get_sales_vs_listings( logger.warning("sales-vs-listings: cannot extract street from %r", address) return _empty() - rows = db.execute( - text( - """ + rows = ( + db.execute( + text( + """ SELECT deal_id, deal_date, deal_price_rub, deal_price_per_m2, deal_area_m2, deal_rooms, deal_floor, deal_address, @@ -1375,21 +1474,27 @@ def get_sales_vs_listings( CAST(:period_months AS integer) ) """ - ), - { - "street_pattern": "%" + street_name + "%", - "area_m2": area_m2, - "rooms": rooms, - "window_days": window_days, - "area_tolerance": area_tolerance, - "period_months": period_months, - }, - ).mappings().all() + ), + { + "street_pattern": "%" + street_name + "%", + "area_m2": area_m2, + "rooms": rooms, + "window_days": window_days, + "area_tolerance": area_tolerance, + "period_months": period_months, + }, + ) + .mappings() + .all() + ) if not rows: logger.info( "sales-vs-listings: no deals street=%r rooms=%d area=%.1f period_months=%d", - street_name, rooms, area_m2, period_months, + street_name, + rooms, + area_m2, + period_months, ) return _empty(reason_street=street_name) @@ -1411,35 +1516,30 @@ def get_sales_vs_listings( int(r["listing_price_rub"]) if r["listing_price_rub"] is not None else None ), listing_price_per_m2=( - int(r["listing_price_per_m2"]) - if r["listing_price_per_m2"] is not None - else None + int(r["listing_price_per_m2"]) if r["listing_price_per_m2"] is not None else None ), listing_area_m2=( float(r["listing_area_m2"]) if r["listing_area_m2"] is not None else None ), days_listing_to_deal=r["days_listing_to_deal"], - discount_pct=( - float(r["discount_pct"]) if r["discount_pct"] is not None else None - ), + discount_pct=(float(r["discount_pct"]) if r["discount_pct"] is not None else None), ) for r in rows ] total_deals = len(pairs) deals_with_listings = sum(1 for p in pairs if p.listing_id is not None) - linkage_rate_pct = ( - round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0 - ) + linkage_rate_pct = round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0 discounts = sorted(p.discount_pct for p in pairs if p.discount_pct is not None) - median_discount = ( - round(_percentile(discounts, 0.5), 2) if discounts else None - ) + median_discount = round(_percentile(discounts, 0.5), 2) if discounts else None logger.info( "sales-vs-listings: street=%r deals=%d with_listings=%d linkage=%.1f%% median_disc=%s", - street_name, total_deals, deals_with_listings, linkage_rate_pct, + street_name, + total_deals, + deals_with_listings, + linkage_rate_pct, f"{median_discount:+.2f}%" if median_discount is not None else "n/a", ) diff --git a/tradein-mvp/backend/tests/test_estimate_idor.py b/tradein-mvp/backend/tests/test_estimate_idor.py new file mode 100644 index 00000000..0e51c2d0 --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimate_idor.py @@ -0,0 +1,256 @@ +"""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