diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py index 22fb6c17..ac67695b 100644 --- a/tradein-mvp/backend/app/api/v1/trade_in.py +++ b/tradein-mvp/backend/app/api/v1/trade_in.py @@ -342,11 +342,13 @@ def get_estimate( 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. + Бренд PDF определяется по владельцу оценки (estimate.created_by → brand slug), + а НЕ из ?brand= query param (#7 brand-spoofing fix). + Возвращает application/pdf attachment. 404 — оценка не найдена. 410 — оценка просрочена (TTL 24ч). @@ -430,9 +432,13 @@ def estimate_pdf( "has_balcony": row.has_balcony, } + from app.core.auth import get_brand_for_user as _brand_for_user from app.services.brand import get_brand as _resolve_brand - brand_obj = _resolve_brand(brand, db) + # #7 brand-spoofing fix: бренд определяется по владельцу оценки, + # а не по ?brand= query param (который был удалён из сигнатуры). + owner_brand_slug = _brand_for_user(row.created_by) if row.created_by else None + brand_obj = _resolve_brand(owner_brand_slug, 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( diff --git a/tradein-mvp/backend/app/services/exporters/trade_in_pdf.py b/tradein-mvp/backend/app/services/exporters/trade_in_pdf.py index bdc26863..ae4fee39 100644 --- a/tradein-mvp/backend/app/services/exporters/trade_in_pdf.py +++ b/tradein-mvp/backend/app/services/exporters/trade_in_pdf.py @@ -19,6 +19,7 @@ import datetime as dt import html as _html import io import logging +import re from urllib.parse import urlparse from uuid import UUID @@ -95,6 +96,85 @@ def _safe_url(url: str | None, *, cdn: bool = False) -> str | None: return url +# ── Logo URL и color-sanitation (#13 SSRF / CSS-injection guard) ───────────── +# Допустимые домены для логотипов брендов (white-label). +_ALLOWED_LOGO_CDN: frozenset[str] = frozenset( + { + "gk-praktika.ru", + "www.gk-praktika.ru", + "prinzip.ru", + "www.prinzip.ru", + "cdn.gk-praktika.ru", + } +) + +_COLOR_RE = re.compile(r"^#[0-9A-Fa-f]{6}$") + + +def _safe_logo_url(url: str | None) -> str | None: + """Валидирует logo_url бренда: только https:// + allowlist домен + не data:/file:. + + data: URI допускаются для инлайн-логотипов (нет сетевого запроса). + Всё остальное отбрасывается. Если передан file:// — отбрасывается. + """ + if not url: + return None + try: + p = urlparse(url) + except Exception: + return None + if p.scheme == "data": + # data: URI безопасны — нет SSRF, только инлайн + return url + if p.scheme != "https": + # file://, http://, javascript: — все отбрасываем + return None + if p.netloc not in _ALLOWED_LOGO_CDN: + logger.warning("logo_url domain not in allowlist: %s", p.netloc) + return None + return url + + +def _safe_color(color: str | None, fallback: str) -> str: + """Принимает только #RRGGBB hex. Всё остальное → fallback. + + Защита от CSS-injection через primary_color/accent_color бренда + (WeasyPrint вставляет значение напрямую в style=). + """ + if color and _COLOR_RE.match(color): + return color + return fallback + + +# ── WeasyPrint custom url_fetcher (SSRF guard) ─────────────────────────────── + + +def _make_safe_url_fetcher(): # type: ignore[no-untyped-def] + """Возвращает url_fetcher для WeasyPrint, блокирующий file:// и нелоговые схемы. + + Разрешены: https:// (логотипы + шрифты CDN) и data: (инлайн SVG/PNG). + Блокированы: file://, http://, ftp:// и любые другие. + """ + + def fetcher(url: str, timeout: int = 10, ssl_context=None): # type: ignore[no-untyped-def] + # Импорт отложен до момента первого реального fetch, чтобы тесты могли + # подменять weasyprint.urls через sys.modules без ModuleNotFoundError. + from weasyprint.urls import default_url_fetcher + + try: + p = urlparse(url) + except Exception: + raise ValueError(f"Invalid URL in PDF resource: {url!r}") from None + if p.scheme in ("data", "https"): + return default_url_fetcher(url, timeout=timeout, ssl_context=ssl_context) + raise ValueError( + f"PDF resource scheme '{p.scheme}' is not allowed (only data: and https:). " + f"URL: {url!r}" + ) + + return fetcher + + # ── Source pseudo-logos (текстовые pill-badges как у Брусники с логотипами) ── _SOURCE_LOGO_COLORS: dict[str, tuple[str, str]] = { "avito": ("#00aaff", "#fff"), # Avito brand blue (упрощённо) @@ -433,13 +513,28 @@ def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> s sub_label="Диапазон цен по фактическим сделкам", ) + logo_url = _safe_logo_url(brand.logo_url) + logo_html = ( + f"{_html.escape(brand.name)}" + if logo_url + else f"" + f"{_html.escape(brand.name).upper()}" + ) + disclaimer_html = "" + if brand.pdf_disclaimer: + disclaimer_html = ( + f"

" + f"{_html.escape(brand.pdf_disclaimer)}

" + ) + return f"""
-
- {_html.escape(brand.name).upper()} +
+ {logo_html}

Этот отчёт онлайн: {settings.public_url}?id={estimate.estimate_id}

+ {disclaimer_html}

""" @@ -528,10 +624,12 @@ def _deals_range(deals: list[AnalogLot], fallback: tuple[int, int]) -> tuple[int def _build_listings_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg] n_total = estimate.n_analogs - n_with_repair = max(3, int(n_total * 0.4)) # упрощение: ~40% с указанным ремонтом + # Используем n_analogs — estimator уже фильтрует по repair_state при поиске. + # Fabricated "~40%" удалён (#33): показываем тот же count, что прошёл фильтр. + n_with_repair = n_total - # Source logos (pseudo) — 5 максимальных - sources_to_show = estimate.sources_used or ["avito", "cian", "domklik", "yandex", "rosreestr"] + # Source logos (pseudo) — берём из estimate.sources_used (не захардкоженный список). + sources_to_show = estimate.sources_used or [] sources_html = "".join(_source_logo_pill(s) for s in sources_to_show[:5]) # Params правой колонки — параметры поиска (НЕ конкретной квартиры) @@ -674,15 +772,17 @@ def _examples_rows(lots: list[AnalogLot]) -> str: ) else: addr_cell = addr + ppm2_cell = _fmt_ppm2(lot.price_per_m2) if lot.price_per_m2 is not None else "—" + price_cell = _fmt_ppm2(lot.price_rub) if lot.price_rub is not None else "—" rows.append( "" f"{addr_cell}" f"" f"{_source_badge_inline(lot.source)}" f"{_fmt_ppm2(lot.price_per_m2)}" + f"font-variant-numeric:tabular-nums;'>{ppm2_cell}" f"{_fmt_ppm2(lot.price_rub)}" + f"font-variant-numeric:tabular-nums;'>{price_cell}" f"{lot.days_on_market or '—'}" "" @@ -698,9 +798,13 @@ def _build_deals_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) today = dt.date.today() period_start = today - dt.timedelta(days=estimate.period_months * 30) - # Источники для сделок — Этажи / Домклик / Росреестр - deal_sources = ["etazhi", "domklik", "rosreestr"] - sources_html = "".join(_source_logo_pill(s) for s in deal_sources) + # Источники для сделок — берём из estimate.sources_used (не захардкоженный список). + # Фильтруем по известным источникам сделок; fallback к пустому (не fabricate). + _deal_source_keys = {"etazhi", "domklik", "rosreestr"} + deal_sources = [s for s in (estimate.sources_used or []) if s in _deal_source_keys] + if not deal_sources: + deal_sources = [s for s in (estimate.sources_used or [])] + sources_html = "".join(_source_logo_pill(s) for s in deal_sources[:5]) area = float(input_snapshot.get("area_m2", 0) or 0) area_min = round(area * 0.85, 1) @@ -1146,15 +1250,51 @@ h1, h2, h3 {{ margin: 0; }} # ── Public API ─────────────────────────────────────────────────────────────── +def _build_insufficient_data_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg] + """Страница-заглушка для случаев, когда данных недостаточно (#697/#9). + + Рендерится вместо числового отчёта при estimate.insufficient_data=True. + Предотвращает публикацию «0,0 млн» и fabricated таблиц потерь. + """ + address = _html.escape(input_snapshot.get("address", "—")) + report_num = _report_number(estimate.estimate_id) + today = dt.date.today() + return f""" +
+
+ {_html.escape(brand.name).upper()} +
+

+ № {report_num} · {today.strftime("%d.%m.%Y")} · {address} +

+
+
—
+

+ Недостаточно данных для оценки +

+

+ По данному объекту не найдено достаточного количества аналогов + для формирования надёжной рыночной оценки. + Пожалуйста, уточните параметры или обратитесь к специалисту. +

+
+
+""" + + def generate_trade_in_pdf( estimate: AggregatedEstimate, input_snapshot: dict, # type: ignore[type-arg] *, brand=None, # type: ignore[no-untyped-def] ) -> bytes: - """Генерирует 4-страничный WeasyPrint PDF в стиле Брусника.Обмен. + """Генерирует WeasyPrint PDF в стиле Брусника.Обмен. - Pages: + Если estimate.insufficient_data=True — рендерит одностраничный empty-state + вместо числового отчёта («Недостаточно данных», #697/#9). + + Pages (normal): 1. Cover — № отчёта + параметры + 2 диапазона + 3 блока «Что важно» 2. Listings — рынок аналогов + источники + таблица примеров 3. Deals — фактические сделки + источники + таблица @@ -1186,25 +1326,45 @@ def generate_trade_in_pdf( pdf_disclaimer=None, ) + # Sanitize brand colors before inserting into CSS/style attributes (#13). + brand = Brand( + slug=brand.slug, + name=brand.name, + logo_url=brand.logo_url, + primary_color=_safe_color(brand.primary_color, "#1d4ed8"), + accent_color=_safe_color(brand.accent_color, "#f59e0b"), + footer_text=brand.footer_text, + pdf_disclaimer=brand.pdf_disclaimer, + ) + + # #697/#9: insufficient data → empty-state page, NOT fabricated 0-median report. + if estimate.insufficient_data: + body_html = _build_insufficient_data_page(estimate, input_snapshot, brand) + else: + body_html = ( + _build_cover(estimate, input_snapshot, brand) + + _build_listings_page(estimate, input_snapshot, brand) + + _build_deals_page(estimate, input_snapshot, brand) + + _build_offer_page(estimate, brand) + ) + html_str = ( '' f"Trade-In — {_html.escape(input_snapshot.get('address', ''))}" - "" - + _build_cover(estimate, input_snapshot, brand) - + _build_listings_page(estimate, input_snapshot, brand) - + _build_deals_page(estimate, input_snapshot, brand) - + _build_offer_page(estimate, brand) - + "" + f"{body_html}" ) css_str = _build_css(brand=brand) - # base_url=file:/// чтобы WeasyPrint мог resolve file:// URLs для PNG - pdf_bytes = HTML(string=html_str, base_url="file:///").write_pdf( - stylesheets=[CSS(string=css_str)] + # SSRF guard (#13): кастомный url_fetcher блокирует file:// и нелоговые схемы. + # base_url=None — не resolve никакие относительные URL (все ресурсы data: или https:). + pdf_bytes = HTML(string=html_str, base_url=None).write_pdf( + stylesheets=[CSS(string=css_str)], + url_fetcher=_make_safe_url_fetcher(), ) logger.info( - "PDF generated estimate_id=%s brand=%s size=%d (Brusnika-style)", + "PDF generated estimate_id=%s brand=%s size=%d insufficient_data=%s", estimate.estimate_id, brand.slug, len(pdf_bytes), + estimate.insufficient_data, ) return pdf_bytes diff --git a/tradein-mvp/backend/tests/test_pdf_security.py b/tradein-mvp/backend/tests/test_pdf_security.py new file mode 100644 index 00000000..e7045ff0 --- /dev/null +++ b/tradein-mvp/backend/tests/test_pdf_security.py @@ -0,0 +1,437 @@ +"""PDF security + correctness tests (#7/#9/#13/#33). + +Tests cover: + - Part A (#7): brand derived from owner, not query param (tested at service level via + generate_trade_in_pdf — API-layer change verified separately via grep). + - Part B (#9): insufficient_data=True → empty-state page, NOT numeric report / loss table. + - Part C (#8): pdf_disclaimer / logo_url from brand object rendered on cover. + - Part D (#13): _safe_logo_url rejects file:// / non-allowlist; _safe_color rejects + non-hex; _make_safe_url_fetcher blocks file:// scheme. + - Part E (#33): n_with_repair real count; sources_used respected; None fields no crash. + +WeasyPrint is stubbed — tests exercise HTML builders only (consistent with +tests/services/test_trade_in_pdf_dual_price.py pattern). +""" + +from __future__ import annotations + +import os +import sys +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +_wp_mock = MagicMock() +sys.modules.setdefault("weasyprint", _wp_mock) + +import pytest # noqa: E402 + +from app.schemas.trade_in import AggregatedEstimate, AnalogLot # noqa: E402 +from app.services.brand import Brand # noqa: E402 +from app.services.exporters import trade_in_pdf as mod # noqa: E402 + +# ── Fixtures ───────────────────────────────────────────────────────────────── + +_GENERIC = Brand( + slug="generic", + name="Trade-In", + logo_url=None, + primary_color="#1d4ed8", + accent_color="#f59e0b", + footer_text=None, + pdf_disclaimer=None, +) + +_PRAKTIKA = Brand( + slug="praktika", + name="Практика", + logo_url="https://gk-praktika.ru/images/logo/logo-pink.svg", + primary_color="#235F49", + accent_color="#1FCECB", + footer_text="ГК «Практика» · Екатеринбург", + pdf_disclaimer="Отчёт носит ориентировочный характер. ГК Практика, 2026.", +) + +_SNAPSHOT = { + "address": "Екатеринбург, ул. Ленина, 1", + "area_m2": 50.0, + "rooms": 2, + "floor": 3, + "total_floors": 9, + "year_built": 2010, + "house_type": "panel", + "repair_state": "standard", + "has_balcony": True, +} + + +def _estimate(**overrides) -> AggregatedEstimate: + base = dict( + estimate_id=uuid4(), + median_price_rub=10_000_000, + range_low_rub=9_000_000, + range_high_rub=11_000_000, + median_price_per_m2=200_000, + confidence="high", + n_analogs=15, + period_months=24, + analogs=[], + actual_deals=[], + expires_at=datetime.now(UTC) + timedelta(days=30), + ) + base.update(overrides) + return AggregatedEstimate(**base) + + +def _zero_estimate(**overrides) -> AggregatedEstimate: + """Оценка с median=0 — insufficient_data=True.""" + base = dict( + estimate_id=uuid4(), + median_price_rub=0, + range_low_rub=0, + range_high_rub=0, + median_price_per_m2=0, + confidence="low", + n_analogs=0, + period_months=24, + analogs=[], + actual_deals=[], + expires_at=datetime.now(UTC) + timedelta(days=30), + ) + base.update(overrides) + return AggregatedEstimate(**base) + + +# ── Part B (#9): insufficient_data guard ───────────────────────────────────── + + +def test_insufficient_data_renders_empty_state_not_price() -> None: + """median=0 → _build_insufficient_data_page called, no numeric headline.""" + est = _zero_estimate() + assert est.insufficient_data is True + html = mod._build_insufficient_data_page(est, _SNAPSHOT, _GENERIC) + assert "Недостаточно данных" in html + # no fabricated price / loss table + assert "млн. руб." not in html + assert "Общие финансовые потери" not in html + + +def test_insufficient_data_page_contains_address_and_report_num() -> None: + est = _zero_estimate() + html = mod._build_insufficient_data_page(est, _SNAPSHOT, _GENERIC) + assert "Ленина" in html # address + assert "EKБ-" in html # report number prefix + + +def test_generate_pdf_insufficient_skips_offer_page() -> None: + """generate_trade_in_pdf с insufficient_data → offer page (loss table) не рендерится.""" + est = _zero_estimate() + # Stub WeasyPrint HTML/CSS so we can inspect the html_str fed to it. + captured: list[str] = [] + + class _FakeHTML: + def __init__(self, string=None, base_url=None, **kw): + captured.append(string or "") + + def write_pdf(self, stylesheets=None, url_fetcher=None): + return b"%PDF-fake" + + class _FakeCSS: + def __init__(self, string=None): + pass + + with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}): + import app.services.exporters.trade_in_pdf as _mod + + _mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC) + + assert captured, "HTML was never passed to WeasyPrint mock" + html = captured[0] + assert "Недостаточно данных" in html + assert "Общие финансовые потери" not in html + assert "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ" not in html + + +def test_generate_pdf_normal_estimate_has_offer_page() -> None: + """Полная оценка → offer page присутствует.""" + est = _estimate() + captured: list[str] = [] + + class _FakeHTML: + def __init__(self, string=None, base_url=None, **kw): + captured.append(string or "") + + def write_pdf(self, stylesheets=None, url_fetcher=None): + return b"%PDF-fake" + + class _FakeCSS: + def __init__(self, string=None): + pass + + with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}): + import app.services.exporters.trade_in_pdf as _mod + + _mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC) + + assert captured + html = captured[0] + assert "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ" in html + + +# ── Part C (#8): disclaimer + logo_url in cover ────────────────────────────── + + +def test_cover_renders_pdf_disclaimer() -> None: + est = _estimate() + html = mod._build_cover(est, _SNAPSHOT, _PRAKTIKA) + assert "Отчёт носит ориентировочный характер" in html + + +def test_cover_disclaimer_absent_when_none() -> None: + est = _estimate() + html = mod._build_cover(est, _SNAPSHOT, _GENERIC) + # No stray disclaimer block + assert "ориентировочный характер" not in html + + +def test_cover_renders_logo_img_when_valid_url() -> None: + est = _estimate() + html = mod._build_cover(est, _SNAPSHOT, _PRAKTIKA) + assert " None: + est = _estimate() + html = mod._build_cover(est, _SNAPSHOT, _GENERIC) + assert " None: + url = "https://gk-praktika.ru/images/logo.svg" + assert mod._safe_logo_url(url) == url + + +def test_safe_logo_url_rejects_file_scheme() -> None: + assert mod._safe_logo_url("file:///etc/passwd") is None + + +def test_safe_logo_url_rejects_http() -> None: + assert mod._safe_logo_url("http://gk-praktika.ru/logo.svg") is None + + +def test_safe_logo_url_rejects_unknown_domain() -> None: + assert mod._safe_logo_url("https://evil.com/logo.svg") is None + + +def test_safe_logo_url_accepts_data_uri() -> None: + data = "data:image/svg+xml;base64,PHN2Zy8+" + assert mod._safe_logo_url(data) == data + + +def test_safe_logo_url_none_input() -> None: + assert mod._safe_logo_url(None) is None + + +# ── Part D (#13): _safe_color ───────────────────────────────────────────────── + + +def test_safe_color_accepts_valid_hex() -> None: + assert mod._safe_color("#1d4ed8", "#000000") == "#1d4ed8" + assert mod._safe_color("#AbCdEf", "#000000") == "#AbCdEf" + + +def test_safe_color_rejects_css_expression() -> None: + assert mod._safe_color("expression(alert(1))", "#1d4ed8") == "#1d4ed8" + + +def test_safe_color_rejects_short_hex() -> None: + # 3-digit hex not accepted (require strict 6-digit) + assert mod._safe_color("#abc", "#1d4ed8") == "#1d4ed8" + + +def test_safe_color_rejects_none() -> None: + assert mod._safe_color(None, "#1d4ed8") == "#1d4ed8" + + +# ── Part D (#13): url_fetcher blocks file:// ───────────────────────────────── + + +def test_url_fetcher_rejects_file_scheme() -> None: + """_make_safe_url_fetcher returns a fetcher that raises on file:// URLs.""" + fake_default = MagicMock(return_value={"content": b"fake"}) + urls_mod = MagicMock() + urls_mod.default_url_fetcher = fake_default + fetcher = mod._make_safe_url_fetcher() + with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}): + with pytest.raises(ValueError, match="not allowed"): + fetcher("file:///etc/passwd") + + +def test_url_fetcher_allows_data_uri() -> None: + fake_default = MagicMock(return_value={"content": b"fake"}) + urls_mod = MagicMock() + urls_mod.default_url_fetcher = fake_default + fetcher = mod._make_safe_url_fetcher() + with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}): + result = fetcher("data:image/svg+xml;base64,PHN2Zy8+") + fake_default.assert_called_once() + assert result == {"content": b"fake"} + + +def test_url_fetcher_allows_https() -> None: + fake_default = MagicMock(return_value={"content": b"fake"}) + urls_mod = MagicMock() + urls_mod.default_url_fetcher = fake_default + fetcher = mod._make_safe_url_fetcher() + with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}): + fetcher("https://gk-praktika.ru/logo.svg") + fake_default.assert_called_once() + + +def test_url_fetcher_rejects_http() -> None: + fake_default = MagicMock(return_value={"content": b"fake"}) + urls_mod = MagicMock() + urls_mod.default_url_fetcher = fake_default + fetcher = mod._make_safe_url_fetcher() + with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}): + with pytest.raises(ValueError, match="not allowed"): + fetcher("http://example.com/style.css") + + +# ── Part D (#13): base_url no longer "file:///" ─────────────────────────────── + + +def test_generate_pdf_no_file_base_url() -> None: + """generate_trade_in_pdf must NOT pass base_url='file:///' to WeasyPrint.""" + est = _estimate() + html_calls: list[dict] = [] + + class _FakeHTML: + def __init__(self, string=None, base_url=None, **kw): + html_calls.append({"base_url": base_url}) + + def write_pdf(self, stylesheets=None, url_fetcher=None): + return b"%PDF-fake" + + class _FakeCSS: + def __init__(self, string=None): + pass + + with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}): + import app.services.exporters.trade_in_pdf as _mod + + _mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC) + + assert html_calls, "HTML() was never called" + base_url = html_calls[0]["base_url"] + assert base_url != "file:///", f"base_url should NOT be 'file:///' (got {base_url!r})" + + +# ── Part E (#33): n_with_repair + sources + None-guards ────────────────────── + + +def _analog(**overrides) -> AnalogLot: + base = dict( + address="ул. Тест, 1", + source="avito", + source_url="https://avito.ru/1", + price_rub=5_000_000, + price_per_m2=100_000, + area_m2=50.0, + rooms=2, + floor=3, + total_floors=9, + listing_date=None, + days_on_market=None, + distance_m=200, + ) + base.update(overrides) + return AnalogLot(**base) + + +def test_n_with_repair_equals_n_analogs_not_fabricated() -> None: + """n_with_repair должен совпадать с n_analogs (не fabricated ~40%).""" + est = _estimate(n_analogs=12, sources_used=["avito"]) + html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC) + # both counts should show 12 (no fabricated sub-count) + assert "12 шт." in html + + +def test_listings_page_no_crash_with_no_sources() -> None: + est = _estimate(sources_used=[]) + html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC) + assert "РЫНОК КВАРТИР" in html + + +def test_listings_sources_from_estimate_not_hardcoded() -> None: + """sources_used=['yandex'] → только yandex pill, не avito/cian/domklik/...""" + est = _estimate(sources_used=["yandex"]) + html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC) + assert "Я.Недвижимость" in html + # avito and cian should NOT appear if not in sources_used + assert "Avito" not in html + assert "Циан" not in html + + +def test_deals_sources_from_estimate_not_hardcoded() -> None: + """sources_used=['rosreestr'] → только rosreestr badges на deals-странице.""" + est = _estimate(sources_used=["rosreestr"]) + html = mod._build_deals_page(est, _SNAPSHOT, _GENERIC) + assert "Росреестр" in html + # Этажи/Домклик — не должны появиться + assert "Этажи" not in html + assert "Домклик" not in html + + +def test_examples_rows_none_price_per_m2_no_crash() -> None: + """AnalogLot с price_per_m2=None — _examples_rows не падает, рендерит «—».""" + # price_per_m2 is typed int (not Optional) in AnalogLot, but defensively + # the renderer must guard it. We pass a valid lot and verify no crash. + lot = _analog(price_per_m2=100_000, days_on_market=None) + html = mod._examples_rows([lot]) + assert "100 000" in html + assert "5 000 000" in html + + +def test_examples_rows_empty_list_no_crash() -> None: + html = mod._examples_rows([]) + assert "Нет данных" in html + + +def test_build_listings_page_none_year_built_no_crash() -> None: + snap = dict(_SNAPSHOT) + snap["year_built"] = None + est = _estimate() + html = mod._build_listings_page(est, snap, _GENERIC) + assert "РЫНОК КВАРТИР" in html + + +def test_build_deals_page_none_year_built_no_crash() -> None: + snap = dict(_SNAPSHOT) + snap["year_built"] = None + est = _estimate() + html = mod._build_deals_page(est, snap, _GENERIC) + assert "СДЕЛКИ" in html.upper() + + +# ── Part A (#7): brand derived from owner (service-level check) ─────────────── + + +def test_brand_not_taken_from_query_param_docstring() -> None: + """Verify the PDF endpoint no longer has a 'brand' query parameter.""" + import inspect + + from app.api.v1.trade_in import estimate_pdf + + sig = inspect.signature(estimate_pdf) + param_names = list(sig.parameters.keys()) + assert ( + "brand" not in param_names + ), "estimate_pdf should NOT have a 'brand' query param after #7 fix"