Light1YT
01a74ade7a
deficit_index pins to -1.0 for every ЕКБ segment (12mo demand flow vs multi-year supply stock → log-ratio clamps) → zero discriminating power, though the oversupply is partly real. Add MOI (gross competing supply / demand_per_mo), the real-estate absorption standard, as an additive non-saturating companion that DISCRIMINATES (Уралмаш 42mo … Чермет 109mo) where deficit cannot. deficit_index math kept exactly as-is (honest absolute: -1 = genuinely oversupplied); docstrings clarify -1 is common and MOI is the discriminating companion (no recalibration). _gross_supply extract-method (single source of truth; _project_supply behavior byte-identical, code-review-verified). Surface MOI in §22 future_market (passthrough) + exec_summary key_numbers/verdict. Guards: no demand → None, no supply → 0. Prod: MOI varies 42→109mo, deficit stays -1. Discrimination test pins MOI separating two segments both at deficit -1. Refs #952.
405 lines
19 KiB
Python
405 lines
19 KiB
Python
"""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,
|
||
_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
|