gendesign/backend/tests/services/exporters/test_excel.py
Light1YT 3cf9fad683
Some checks failed
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 7s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 6m38s
CI / backend-tests (pull_request) Successful in 6m37s
Deploy / build-backend (push) Blocked by required conditions
Deploy / build-worker (push) Blocked by required conditions
Deploy / build-frontend (push) Blocked by required conditions
Deploy / deploy (push) Blocked by required conditions
Deploy / changes (push) Has been cancelled
fix(backend): экранировать Excel formula-injection (#1244) + увести chat-чтение с event loop (#1245)
#1244 (security): внешние/скрейпинг-строки (comm_name из DOM.РФ, headline/usp_text)
с ведущим = + - @ \t \r писались как есть → openpyxl сохранял как формулы
(data_type='f'), исполнялись при открытии в Excel/LibreOffice. _sanitize_formula
префиксует такие строки апострофом (OWASP CSV-injection escape); числа/даты/bool
не трогаются. _write_kv labels тоже санитизируются. Подтверждено на openpyxl 3.1.5.

#1245 (concurrency): async ask() вызывал sync get_report_for_chat() (sync SQLAlchemy
тянет крупный JSONB §22-отчёт) напрямую — блокировал event loop, в отличие от
LLM-ветки (run_in_threadpool). Обёрнуто в run_in_threadpool.

Closes #1244
Closes #1245
2026-06-13 18:10:21 +05:00

475 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Unit-тесты §13 Excel-экспортёра (#991, 955-A5) — `export_report_xlsx`.
Чистые тесты БЕЗ БД/сети (экспортёр только ПОТРЕБЛЯЕТ уже-собранный отчёт):
• полный `SiteFinderReport` → непустые .xlsx-байты, перечитываемые openpyxl;
• все шесть ожидаемых листов присутствуют (Сводка / Рынок сейчас / Будущий рынок /
Продукт ТЗ / Сценарии / Скоринг);
• известные ячейки заполнены (заголовок вердикта, класс продукта, ADVISORY-маркер);
• экспортёр принимает КАК dataclass-инстанс, ТАК и его `as_dict()`-словарь;
• частичный / пустой отчёт → всё равно валидный .xlsx (graceful, без падения).
Детерминированно, без LLM. DATABASE_URL выставляем до импорта app-модулей (зеркало
test_report.py) — на случай side-effect'ов импорта пакета.
"""
from __future__ import annotations
import os
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import datetime as dt
from io import BytesIO
from openpyxl import load_workbook
from app.services.exporters.excel import (
_SHEET_FUTURE_MARKET,
_SHEET_MARKET_NOW,
_SHEET_PRODUCT_TZ,
_SHEET_SCENARIOS,
_SHEET_SCORING,
_SHEET_SUMMARY,
_fmt,
_sanitize_formula,
_scenario_deficit_index,
export_report_xlsx,
)
from app.services.forecasting.demand_supply_forecast import DemandSupplyForecast
from app.services.forecasting.report import (
ReportConfidence,
ReportExecSummary,
ReportFutureMarket,
ReportMarketNow,
ReportMeta,
ReportProductTz,
ReportScenarios,
ReportScoring,
SiteFinderReport,
)
from app.services.forecasting.scenarios import ScenarioForecast
def _forecast(horizon: int, demand: float, supply: float, deficit: float) -> dict[str, object]:
"""Реальный `DemandSupplyForecast.as_dict()` — ключи гарантированно из контракта #952.
Строим frozen-dataclass БЕЗ БД и сериализуем через его собственный `as_dict()`: так
тест не может разойтись с реальной формой (projected_demand_units / projected_supply_
units), как это случилось со старой hand-typed фикстурой (demand/supply — #989/#991).
"""
return DemandSupplyForecast(
segment={"obj_class": "комфорт"},
horizon_months=horizon,
base_pace_units_per_mo=8.0,
demand_norm_coefficient=1.0,
macro_coefficient=1.0,
projected_demand_units=demand,
open_units=300,
hidden_release_units=80.0,
future_online_units=20.0,
projected_supply_units=supply,
balance_units=demand - supply,
balance_ratio=(demand / supply if supply else None),
deficit_index=deficit,
months_of_inventory=(supply / (demand / horizon) if demand else None),
rate_future=18.0,
rate_sensitivity_phrase="спрос эластичен к ставке",
future_competitors=[{"obj_id": 3, "comm_name": "ЖК Гамма"}],
advisory=True,
confidence="medium",
).as_dict()
def _scenario(name: str, deficit_12mo: float) -> dict[str, object]:
"""Реальный `ScenarioForecast.as_dict()` — ключи из контракта #984 (НЕ hand-typed).
У сценария НЕТ скалярного «overall»: есть `forecasts` (по горизонтам) с deficit_index.
Экспортёр тянет сводный дефицит 12-мес горизонта — здесь даём именно его.
"""
return ScenarioForecast(
scenario=name, # type: ignore[arg-type] # Literal в проде; в тесте — str-имя
rate_path={6: 18.0, 12: 18.0},
forecasts=[
DemandSupplyForecast(
segment={"obj_class": "комфорт"},
horizon_months=12,
base_pace_units_per_mo=8.0,
demand_norm_coefficient=1.0,
macro_coefficient=1.0,
projected_demand_units=920.0,
open_units=300,
hidden_release_units=80.0,
future_online_units=20.0,
projected_supply_units=700.0,
balance_units=220.0,
balance_ratio=1.314,
deficit_index=deficit_12mo,
months_of_inventory=9.1,
rate_future=18.0,
rate_sensitivity_phrase=None,
future_competitors=[],
advisory=True,
confidence="medium",
)
],
advisory=True,
).as_dict()
# Шесть ожидаемых листов (по одному на содержательную секцию §13).
_EXPECTED_SHEETS: tuple[str, ...] = (
_SHEET_SUMMARY,
_SHEET_MARKET_NOW,
_SHEET_FUTURE_MARKET,
_SHEET_PRODUCT_TZ,
_SHEET_SCENARIOS,
_SHEET_SCORING,
)
def _full_report() -> SiteFinderReport:
"""Полный SiteFinderReport из реалистичных sample-данных (форма #987 `as_dict()`)."""
return SiteFinderReport(
exec_summary=ReportExecSummary(
headline="Строить комфорт-класс, акцент на студии и 1-к.",
verdict="Дефицит малых форматов на горизонте 12 мес; рынок поглощает быстро.",
key_numbers={"overall_score": 0.71, "deficit_index": 0.34, "horizon_months": 12},
overall_confidence="medium",
),
market_now=ReportMarketNow(
market_metrics={"unit_velocity": 8.2, "overstock_index": 0.21, "confidence": "medium"},
competitors=[
{"obj_id": 1, "comm_name": "ЖК Альфа", "relevance_weight": 0.8},
{"obj_id": 2, "comm_name": "ЖК Бета", "relevance_weight": 0.5},
],
supply_layers={"open_units": 1200, "hidden_units": 800},
summary="Текущий рынок: умеренная абсорбция, средняя плотность конкурентов.",
),
future_market=ReportFutureMarket(
forecasts_by_horizon=[
_forecast(6, demand=410.0, supply=380.0, deficit=0.2),
_forecast(12, demand=920.0, supply=700.0, deficit=0.34),
],
future_supply={
"district": "Верх-Исетский",
"horizon_months": 12,
"premise_kind": "квартира",
"confidence": "low",
"index": 0.42,
"breakdown": {
"open_units": 1200,
"hidden_units": 800,
"future_units_by_horizon": 450.0,
"monthly_absorption_units": 8.2,
"months_of_pressure": 9.0,
"index": 0.42,
},
},
future_competitors=[{"obj_id": 3, "stage": "котлован"}],
scenarios_summary={"base": 0.34, "conservative": 0.18, "aggressive": 0.49},
summary="Будущий рынок: дефицит сохраняется на 12 мес.",
),
product_tz=ReportProductTz(
obj_class="комфорт",
mix=[
{"bucket": "1-Студия", "pct": 35},
{"bucket": "2-1-к", "pct": 40},
{"bucket": "3-2-к", "pct": 25},
],
commercial={"available": False, "caveat": "коммерция: нет достаточных данных"},
usp=[{"segment": "1-Студия", "usp_text": "Дефицит студий — стройте их."}],
reasons=[{"why": "deficit_index +0.34 на горизонте 12 мес", "advisory": True}],
summary="Рекомендован комфорт-класс с акцентом на малые форматы.",
),
scenarios=ReportScenarios(
by_scenario={
"conservative": _scenario("conservative", deficit_12mo=0.18),
"base": _scenario("base", deficit_12mo=0.34),
"aggressive": _scenario("aggressive", deficit_12mo=0.49),
},
summary="Разброс сценариев умеренный.",
),
scoring=ReportScoring(
product_scores={
"overall": 0.71,
"scores": {"market_fit": {"value": 0.67}, "demand": {"value": 0.51}},
"advisory": True,
},
special_indices={
"indices": {"launch_window": {"value": 0.6, "label": "12 мес"}},
"advisory": True,
},
overall=0.71,
),
confidence=ReportConfidence(
level="medium",
rationale="Источники advisory-capped; данные средней плотности.",
factors={"data_coverage": 0.6, "engine_validated": False},
),
meta=ReportMeta(
cad_num="66:41:0000000:1",
district="Верх-Исетский",
segment={"obj_class": "комфорт", "room_bucket": "1-к 30-45"},
horizons=[6, 12, 18, 24],
generated_at=dt.date(2026, 6, 3),
),
)
def _reload(payload: bytes): # type: ignore[no-untyped-def]
"""Перечитать .xlsx-байты обратно в openpyxl Workbook (round-trip-проверка)."""
return load_workbook(BytesIO(payload))
# ── Полный отчёт: непустые байты, перечитываются, листы + ячейки на месте ──────
class TestFullReportExport:
def test_returns_non_empty_bytes(self) -> None:
payload = export_report_xlsx(_full_report())
assert isinstance(payload, bytes)
assert len(payload) > 0
def test_openpyxl_can_reload(self) -> None:
# Главный контракт: байты — валидный .xlsx, перечитываемый openpyxl.
wb = _reload(export_report_xlsx(_full_report()))
assert wb.sheetnames # не пусто
def test_expected_sheet_titles_present(self) -> None:
wb = _reload(export_report_xlsx(_full_report()))
for title in _EXPECTED_SHEETS:
assert title in wb.sheetnames, f"отсутствует лист {title}"
def test_known_cells_populated_summary(self) -> None:
wb = _reload(export_report_xlsx(_full_report()))
ws = wb[_SHEET_SUMMARY]
# Заголовок книги + ADVISORY-маркер в первых строках.
assert "§13" in str(ws.cell(row=1, column=1).value)
assert "ADVISORY" in str(ws.cell(row=2, column=1).value)
# Заголовок вердикта (значение метки «Заголовок») присутствует где-то на листе.
flat = [str(c.value) for col in ws.iter_cols() for c in col if c.value is not None]
assert any("Строить комфорт" in v for v in flat)
assert any("66:41:0000000:1" in v for v in flat)
def test_known_cells_populated_product_tz(self) -> None:
wb = _reload(export_report_xlsx(_full_report()))
ws = wb[_SHEET_PRODUCT_TZ]
flat = [str(c.value) for col in ws.iter_cols() for c in col if c.value is not None]
assert any("комфорт" in v for v in flat)
assert any("1-Студия" in v for v in flat)
def test_known_cells_populated_scoring(self) -> None:
wb = _reload(export_report_xlsx(_full_report()))
ws = wb[_SHEET_SCORING]
flat = [c.value for col in ws.iter_cols() for c in col if c.value is not None]
# overall-скор 0.71 присутствует как число (округление сохраняет значение).
assert 0.71 in flat
assert any(isinstance(v, str) and "market_fit" in v for v in flat)
def test_accepts_as_dict_input(self) -> None:
# Экспортёр принимает уже-словарь `as_dict()` (не только dataclass).
payload = export_report_xlsx(_full_report().as_dict())
wb = _reload(payload)
for title in _EXPECTED_SHEETS:
assert title in wb.sheetnames
# ── Regression #989/#991: реальные ключи контракта попадают в ячейки (НЕ "—") ───
# Тянем значения из РЕАЛЬНОГО `SiteFinderReport.as_dict()` (а не hand-typed строк):
# если экспортёр снова прочитает несуществующий ключ (demand/supply / per-scenario
# overall — как до фикса), число пропадёт из ячеек → тест упадёт. Контракт и тест
# не разойдутся молча (root-cause старого зелёного теста — устаревшая фикстура).
def _cell_values(ws) -> list: # type: ignore[no-untyped-def]
"""Все непустые значения ячеек листа (как есть, без str-приведения)."""
return [c.value for col in ws.iter_cols() for c in col if c.value is not None]
class TestContractKeysWritten:
def test_demand_supply_units_in_cells_not_dash(self) -> None:
data = _full_report().as_dict()
forecasts = data["future_market"]["forecasts_by_horizon"]
assert forecasts, "фикстура должна содержать прогнозы по горизонтам"
ws = _reload(export_report_xlsx(_full_report()))[_SHEET_FUTURE_MARKET]
values = _cell_values(ws)
for f in forecasts:
demand = f["projected_demand_units"]
supply = f["projected_supply_units"]
assert demand is not None and supply is not None
assert _fmt(demand) in values, f"спрос {demand} не в ячейках (ключ читается?)"
assert _fmt(supply) in values, f"предложение {supply} не в ячейках"
# «—» НЕ должен стоять там, где число существует.
assert _dash_count_in_forecast_table(ws) == 0
def test_scenario_deficit_in_cells_not_dash(self) -> None:
data = _full_report().as_dict()
by_scenario = data["scenarios"]["by_scenario"]
assert by_scenario, "фикстура должна содержать сценарии"
ws = _reload(export_report_xlsx(_full_report()))[_SHEET_SCENARIOS]
values = _cell_values(ws)
for payload in by_scenario.values():
di = _scenario_deficit_index(payload)
assert di is not None
assert _fmt(di) in values, f"дефицит сценария {di} не в ячейках"
def test_overall_score_in_cells(self) -> None:
data = _full_report().as_dict()
overall = data["scoring"]["overall"]
assert overall is not None
ws = _reload(export_report_xlsx(_full_report()))[_SHEET_SCORING]
assert _fmt(overall) in _cell_values(ws)
def test_future_supply_section_in_cells(self) -> None:
# #991: давление будущего предложения (index + breakdown по слоям) на листе.
data = _full_report().as_dict()
fs = data["future_market"]["future_supply"]
assert fs is not None
ws = _reload(export_report_xlsx(_full_report()))[_SHEET_FUTURE_MARKET]
flat_str = [str(v) for v in _cell_values(ws)]
values = _cell_values(ws)
assert any("Давление будущего предложения" in v for v in flat_str)
assert _fmt(fs["index"]) in values
assert _fmt(fs["breakdown"]["open_units"]) in values
def test_confidence_factors_in_cells(self) -> None:
# #991: confidence.factors (ключ→значение) на листе «Сводка».
ws = _reload(export_report_xlsx(_full_report()))[_SHEET_SUMMARY]
flat_str = [str(v) for v in _cell_values(ws)]
assert any("Факторы уверенности" in v for v in flat_str)
assert any("data_coverage" in v for v in flat_str)
def _dash_count_in_forecast_table(ws) -> int: # type: ignore[no-untyped-def]
"""Сколько ячеек спроса/предложения в таблице прогноза равны «—» (должно быть 0).
Находим строку-шапку «Спрос»/«Предложение», считаем «—» в этих колонках ниже неё.
"""
from app.services.exporters.excel import _DASH
header_row = None
demand_col = supply_col = None
for r in range(1, ws.max_row + 1):
for c in range(1, ws.max_column + 1):
val = ws.cell(row=r, column=c).value
if val == "Спрос":
header_row, demand_col = r, c
elif val == "Предложение":
supply_col = c
if header_row is None or demand_col is None or supply_col is None:
return 0
count = 0
for r in range(header_row + 1, ws.max_row + 1):
for col in (demand_col, supply_col):
if ws.cell(row=r, column=col).value == _DASH:
count += 1
return count
# ── Graceful: частичный / пустой / мусорный вход → валидный .xlsx без падения ──
class TestGracefulPartialReport:
def test_empty_report_still_valid_xlsx(self) -> None:
# Полностью дефолтный (пустой) отчёт — валидный .xlsx, все листы, не падает.
payload = export_report_xlsx(SiteFinderReport())
assert len(payload) > 0
wb = _reload(payload)
for title in _EXPECTED_SHEETS:
assert title in wb.sheetnames
def test_empty_report_shows_no_data_marker(self) -> None:
wb = _reload(export_report_xlsx(SiteFinderReport()))
ws = wb[_SHEET_SCORING]
flat = [str(c.value) for col in ws.iter_cols() for c in col if c.value is not None]
# Пустая секция рисует заглушку «нет данных» (graceful), а не падает.
assert any("нет данных" in v for v in flat)
def test_partial_report_some_sections(self) -> None:
# Заполнены только meta + exec_summary — остальные пусты, .xlsx валиден.
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Тонкий анализ — только заголовок."),
meta=ReportMeta(cad_num="66:41:0000000:2", horizons=[12]),
)
wb = _reload(export_report_xlsx(report))
ws = wb[_SHEET_SUMMARY]
flat = [str(c.value) for col in ws.iter_cols() for c in col if c.value is not None]
assert any("Тонкий анализ" in v for v in flat)
assert any("66:41:0000000:2" in v for v in flat)
def test_garbage_input_does_not_crash(self) -> None:
# Мусор (None / не-отчёт) → пустой, но валидный .xlsx (нормализация в {}).
for junk in (None, 123, "not a report"):
payload = export_report_xlsx(junk)
assert len(payload) > 0
wb = _reload(payload)
assert _SHEET_SUMMARY in wb.sheetnames
# ── #1244: formula/DDE-injection — внешние строки не должны стать формулой Excel ──
# openpyxl сохраняет строку с лидирующим `= + - @ \t \r` как data_type='f' (формула),
# Excel/LibreOffice исполнит её при открытии. Внешние/скрейпинг-поля (comm_name из
# DOM.РФ, headline/usp_text/advisory) попадают в ячейки через `_fmt` → должны быть
# нейтрализованы префиксом `'` (OWASP CSV-injection) и трактоваться как ТЕКСТ.
class TestFormulaInjection:
def test_sanitize_prefixes_dangerous_strings(self) -> None:
# Каждый OWASP-префикс получает ведущий апостроф.
for payload in ("=SUM(A1:A2)", "+1+1", "-2+3", "@cmd", "\tTAB", "\rCR"):
assert _sanitize_formula(payload) == "'" + payload
def test_sanitize_leaves_safe_strings_untouched(self) -> None:
for safe in ("ЖК Альфа", "комфорт", "250 000", "0.34", ""):
assert _sanitize_formula(safe) == safe
def test_fmt_escapes_formula_string(self) -> None:
# Строка-«формула» через _fmt → текст с апострофом (не исполняемая формула).
assert _fmt("=1+1") == "'=1+1"
assert _fmt("@SUM(1)") == "'@SUM(1)"
def test_fmt_does_not_touch_numbers_or_safe_text(self) -> None:
# Числа/безопасный текст не трогаем: отрицательный float — это значение, не строка.
assert _fmt(-0.5) == -0.5
assert _fmt(-3) == -3
assert _fmt("ЖК Бета") == "ЖК Бета"
def test_injected_competitor_name_written_as_text_not_formula(self) -> None:
# comm_name из скрейпинга DOM.РФ с лидирующим `=` → ячейка text, не формула.
report = SiteFinderReport(
market_now=ReportMarketNow(
competitors=[
{"obj_id": 1, "comm_name": "=cmd|'/c calc'!A1", "relevance_weight": 0.8}
],
),
)
ws = _reload(export_report_xlsx(report))[_SHEET_MARKET_NOW]
injected = [
c
for col in ws.iter_cols()
for c in col
if isinstance(c.value, str) and "calc" in c.value
]
assert injected, "имя конкурента должно присутствовать в книге"
for cell in injected:
assert cell.data_type == "s", "ячейка должна быть текстом, не формулой ('f')"
assert cell.value.startswith("'="), "опасный префикс должен быть экранирован"
def test_injected_headline_and_usp_written_as_text(self) -> None:
# headline (Сводка) и usp_text (Продукт ТЗ) — тоже внешние → не формула.
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="=HYPERLINK('http://evil','x')"),
product_tz=ReportProductTz(
usp=[{"segment": "1-Студия", "usp_text": "+evil()"}],
),
)
payload = export_report_xlsx(report)
wb = _reload(payload)
formula_cells = [
c
for ws in wb.worksheets
for col in ws.iter_cols()
for c in col
if c.data_type == "f"
]
assert not formula_cells, f"ни одна ячейка не должна быть формулой: {formula_cells}"