gendesign/backend/tests/services/forecasting/test_report.py
Light1YT 6ffa34eed6 feat(forecasting): §13 SiteFinderReport structural object (#987, 955-A1)
Add frozen, JSON-serializable SiteFinderReport container with 8 nested section
dataclasses (exec_summary, market_now, future_market, product_tz, scenarios,
scoring, confidence, meta). Pure container — no DB/compute/LLM; populated by
#988 assembler, consumed by #989 exporters + #957 chat. All fields
optional/defaulted (partial report valid); as_dict() JSON-safe (dates→iso);
advisory=True; schema_version "1.0". 14 unit tests.
2026-06-03 13:32:56 +05:00

265 lines
13 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 SiteFinderReport (#987, 955-A1) — структурный контейнер отчёта.
Чистые тесты БЕЗ БД (объект ничего не считает — это PURE-контейнер):
• полный отчёт из sample-данных всех восьми секций → `as_dict()` JSON-сериализуем
(`json.dumps` проходит), ключевые поля round-trip'ят, все восемь секций присутствуют;
• частичный / пустой отчёт ВАЛИДЕН (graceful) и тоже JSON-сериализуем;
• schema_version присутствует (на верхнем уровне и в meta) и == _SCHEMA_VERSION;
• advisory ВСЕГДА True (на отчёте и в meta.generated_advisory);
• даты в meta нормализуются в ISO-строки (JSON-safe, без утечки `date`);
• под-сервис-dict'ы держатся как есть (отчёт их НЕ пере-деривирует).
Детерминированно, без LLM, без сети. DATABASE_URL выставляем до импорта app-модулей
(зеркало test_scenarios.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
import json
from typing import Any
from app.services.forecasting.report import (
_SCHEMA_VERSION,
ReportConfidence,
ReportExecSummary,
ReportFutureMarket,
ReportMarketNow,
ReportMeta,
ReportProductTz,
ReportScenarios,
ReportScoring,
SiteFinderReport,
)
# Восемь обязательных секций §13 — стабильный контракт `as_dict()`.
_SECTION_KEYS: tuple[str, ...] = (
"exec_summary",
"market_now",
"future_market",
"product_tz",
"scenarios",
"scoring",
"confidence",
"meta",
)
def _full_report() -> SiteFinderReport:
"""Полный SiteFinderReport из реалистичных sample-данных всех восьми секций.
Sample-dict'ы зеркалят форму `as_dict()` под-сервисов (плоские, JSON-safe) — не
зовём живые сервисы (контейнер их не считает). meta.generated_at — `date`, чтобы
проверить ISO-нормализацию.
"""
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=[
{"horizon_months": 6, "deficit_index": 0.2, "confidence": "medium"},
{"horizon_months": 12, "deficit_index": 0.34, "confidence": "medium"},
],
future_supply={"index": 0.42, "horizon_months": 12, "confidence": "low"},
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": {"overall": 0.18, "advisory": True},
"base": {"overall": 0.34, "advisory": True},
"aggressive": {"overall": 0.49, "advisory": True},
},
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),
),
)
# ── Полный отчёт: JSON-serializability + round-trip + восемь секций ────────────
class TestFullReportSerialization:
def test_as_dict_is_json_serializable(self) -> None:
# Главный контракт: as_dict() проходит json.dumps без default= (ничего сырого).
payload = _full_report().as_dict()
dumped = json.dumps(payload, ensure_ascii=False)
assert isinstance(dumped, str)
# И обратно парсится в эквивалентный dict (полный round-trip через JSON).
assert json.loads(dumped) == payload
def test_all_eight_sections_present(self) -> None:
payload = _full_report().as_dict()
for key in _SECTION_KEYS:
assert key in payload, f"отсутствует секция {key}"
# Каждая секция сериализована в dict (не dataclass — без утечки объектов).
for key in _SECTION_KEYS:
assert isinstance(payload[key], dict)
def test_key_fields_round_trip(self) -> None:
payload = _full_report().as_dict()
assert payload["exec_summary"]["headline"].startswith("Строить комфорт")
assert payload["exec_summary"]["overall_confidence"] == "medium"
assert payload["product_tz"]["obj_class"] == "комфорт"
assert payload["scoring"]["overall"] == 0.71
assert payload["confidence"]["level"] == "medium"
assert payload["meta"]["cad_num"] == "66:41:0000000:1"
assert payload["meta"]["horizons"] == [6, 12, 18, 24]
def test_nested_subservice_dicts_pass_through(self) -> None:
# Контейнер НЕ пере-деривирует под-выводы — держит их как есть.
payload = _full_report().as_dict()
assert payload["market_now"]["market_metrics"]["unit_velocity"] == 8.2
assert len(payload["market_now"]["competitors"]) == 2
assert payload["scoring"]["product_scores"]["scores"]["market_fit"]["value"] == 0.67
assert payload["future_market"]["forecasts_by_horizon"][1]["deficit_index"] == 0.34
def test_date_normalized_to_iso(self) -> None:
# meta.generated_at (date) → ISO-строка (JSON-safe, без утечки date).
payload = _full_report().as_dict()
assert payload["meta"]["generated_at"] == "2026-06-03"
assert isinstance(payload["meta"]["generated_at"], str)
# ── schema_version + advisory ─────────────────────────────────────────────────
class TestSchemaVersionAndAdvisory:
def test_schema_version_present_top_level_and_meta(self) -> None:
payload = _full_report().as_dict()
assert payload["schema_version"] == _SCHEMA_VERSION
assert payload["meta"]["schema_version"] == _SCHEMA_VERSION
def test_advisory_always_true(self) -> None:
report = _full_report()
assert report.advisory is True
payload = report.as_dict()
assert payload["advisory"] is True
assert payload["meta"]["generated_advisory"] is True
def test_empty_report_advisory_still_true(self) -> None:
# Дефолтный (пустой) отчёт тоже советующий.
payload = SiteFinderReport().as_dict()
assert payload["advisory"] is True
assert payload["meta"]["generated_advisory"] is True
# ── Graceful: частичный / пустой отчёт валиден ────────────────────────────────
class TestPartialAndEmptyReport:
def test_empty_report_is_valid_and_json_serializable(self) -> None:
# Полностью дефолтный отчёт — валиден, восемь секций, JSON-сериализуем.
payload = SiteFinderReport().as_dict()
for key in _SECTION_KEYS:
assert key in payload
dumped = json.dumps(payload, ensure_ascii=False)
assert json.loads(dumped) == payload
def test_empty_sections_have_safe_defaults(self) -> None:
payload = SiteFinderReport().as_dict()
# Скаляры → None, коллекции → пустые (не падаем, не None-вместо-списка).
assert payload["exec_summary"]["headline"] is None
assert payload["exec_summary"]["key_numbers"] == {}
assert payload["market_now"]["competitors"] == []
assert payload["product_tz"]["mix"] == []
assert payload["scenarios"]["by_scenario"] == {}
assert payload["scoring"]["overall"] is None
assert payload["confidence"]["level"] is None
assert payload["meta"]["horizons"] == []
assert payload["meta"]["generated_at"] is None
def test_partial_report_only_some_sections(self) -> None:
# Заполнена только meta + exec_summary — остальные дефолтны, отчёт валиден.
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Тонкий анализ — только заголовок."),
meta=ReportMeta(cad_num="66:41:0000000:2", horizons=[12]),
)
payload = report.as_dict()
assert payload["exec_summary"]["headline"] == "Тонкий анализ — только заголовок."
assert payload["meta"]["cad_num"] == "66:41:0000000:2"
# Незаполненные секции присутствуют и пусты (graceful).
assert payload["future_market"]["forecasts_by_horizon"] == []
assert payload["scoring"]["product_scores"] is None
# И всё ещё JSON-сериализуемо.
json.dumps(payload, ensure_ascii=False)
def test_iso_passthrough_for_string_and_datetime(self) -> None:
# generated_at принимает date / datetime / готовую ISO-строку — все → str.
as_str = SiteFinderReport(meta=ReportMeta(generated_at="2026-06-03T10:00:00")).as_dict()
assert as_str["meta"]["generated_at"] == "2026-06-03T10:00:00"
as_dtm = SiteFinderReport(
meta=ReportMeta(generated_at=dt.datetime(2026, 6, 3, 10, 0, 0))
).as_dict()
assert as_dtm["meta"]["generated_at"].startswith("2026-06-03T10:00:00")
# ── Иммутабельность (frozen dataclass — зеркало соседей) ───────────────────────
class TestFrozen:
def test_report_is_frozen(self) -> None:
import dataclasses
report = SiteFinderReport()
with __import__("pytest").raises(dataclasses.FrozenInstanceError):
report.advisory = False # type: ignore[misc]
def test_as_dict_returns_plain_jsonable_types(self) -> None:
# Никакой утечки dataclass'ов: все значения секций — dict/list/str/num/bool/None.
payload: dict[str, Any] = _full_report().as_dict()
for value in payload.values():
assert isinstance(value, (dict, list, str, int, float, bool, type(None)))