"""Unit-тесты §13 сборщика отчёта (#988, 955-A2) — ЧИСТОЕ маппирование, БЕЗ БД. Сборщик PURE (берёт уже-посчитанные входы, в БД не ходит, §9.x не пересобирает) → тесты чистые, без фикстур БД: • полный сбор (sample analyze dict + sample advisory dict'ы всех под-сервисов) → SiteFinderReport со ВСЕМИ восемью секциями заполненными, `as_dict()` JSON-сериализуем, confidence посчитана через #990 (структурная rationale присутствует), exec_summary headline непустой, advisory True; • частичный сбор (ТОЛЬКО analyze dict) → валидный частичный отчёт (market_now заполнен, future-секции пусты, confidence 'low'); • вход КАК dataclass-инстанс, ТАК и как `.as_dict()`-словарь — оба нормализуются (`_as_dict_or`), отчёт идентичен; • извлечение сигналов качества данных для #990 (deal_count/analog_count/domrf_coverage/ history_months/confounded) + exec_summary-синтез; • graceful: пустой analyze {} → отчёт всё равно валиден (8 секций, JSON-safe). Детерминированно, без LLM, без сети. DATABASE_URL до импорта app-модулей (зеркало соседних тестов — на случай side-effect'ов импорта пакета forecasting). """ from __future__ import annotations import json import os from dataclasses import dataclass from typing import Any os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") from app.services.forecasting.report import SiteFinderReport from app.services.forecasting.report_assembler import ( _analog_count, _as_dict_or, _component_confidences, _confounded, _deal_count, _domrf_coverage, _history_months, _primary_deficit_index, _primary_months_of_inventory, assemble_report, ) # Восемь обязательных секций §13 — стабильный контракт `as_dict()`. _SECTION_KEYS: tuple[str, ...] = ( "exec_summary", "market_now", "future_market", "product_tz", "scenarios", "scoring", "confidence", "meta", ) # ────────────────────────────────────────────────────────────────────────────── # Sample-данные: форма `as_dict()` под-сервисов (плоские, JSON-safe). НЕ зовём живые # сервисы (сборщик их не считает — берёт готовые входы). # ────────────────────────────────────────────────────────────────────────────── def _sample_analyze() -> dict[str, Any]: """Реалистичный (усечённый) dict вывода analyze_parcel — только релевантные ключи.""" return { "cad_num": "66:41:0000000:1", "district": {"district_name": "Верх-Исетский"}, "competitors": [ {"obj_id": 1, "comm_name": "ЖК Альфа", "distance_m": 320.0}, {"obj_id": 2, "comm_name": "ЖК Бета", "distance_m": 540.0}, ], "market_pulse": { "competitors_total": 12, "competitors_with_price": 9, "coverage_pct": 75.0, "market_avg_price_per_m2": 145000, }, "market_avg_price_per_m2": 145000, "market_data_coverage_pct": 75.0, "poi_count": 18, } def _sample_market_metrics() -> dict[str, Any]: """Форма MarketMetrics.as_dict() (#950, §9.2) — несёт счётчики качества данных.""" return { "district": "Верх-Исетский", "obj_count": 7, "n_lots": 240, "n_sold": 88, "n_available": 152, "window_months": 18, "premise_kind": "квартира", "confidence": "medium", "unit_velocity": 8.2, "overstock_index": 0.21, "absorption_rate": 0.05, } def _sample_supply_layers() -> dict[str, Any]: """Форма supply-слоёв (#950, §9.3) — несёт domrf_coverage.""" return {"open_units": 1200, "hidden_units": 800, "domrf_coverage": 0.55} def _sample_forecasts() -> list[dict[str, Any]]: """Per-горизонт DemandSupplyForecast.as_dict() (#952) — деривируем дефицит/конкурентов.""" return [ { "horizon_months": 6, "deficit_index": 0.21, "months_of_inventory": 18.5, "confidence": "medium", "future_competitors": [], "confounded": False, }, { "horizon_months": 12, "deficit_index": 0.34, "months_of_inventory": 24.2, "confidence": "medium", "future_competitors": [{"obj_id": 3, "comm_name": "ЖК Гамма", "relevance_weight": 0.6}], "confounded": False, }, { "horizon_months": 18, "deficit_index": 0.28, "months_of_inventory": 31.0, "confidence": "low", "future_competitors": [], "confounded": False, }, { "horizon_months": 24, "deficit_index": 0.19, "months_of_inventory": 40.7, "confidence": "low", "future_competitors": [], "confounded": False, }, ] def _sample_future_supply() -> dict[str, Any]: """Форма FutureSupplyPressure.as_dict() (#950, §9.3).""" return { "district": "Верх-Исетский", "horizon_months": 12, "confidence": "low", "index": 0.42, "breakdown": {"open_units": 1200, "hidden_units": 800}, } def _sample_scenarios() -> list[dict[str, Any]]: """Три ScenarioForecast.as_dict() (#984) — conservative/base/aggressive.""" return [ { "scenario": "conservative", "advisory": True, "rate_path": {12: 18.0}, "forecasts": [{"horizon_months": 12, "deficit_index": 0.18, "confidence": "medium"}], }, { "scenario": "base", "advisory": True, "rate_path": {12: 16.0}, "forecasts": [{"horizon_months": 12, "deficit_index": 0.34, "confidence": "medium"}], }, { "scenario": "aggressive", "advisory": True, "rate_path": {12: 13.0}, "forecasts": [{"horizon_months": 12, "deficit_index": 0.49, "confidence": "medium"}], }, ] def _sample_overlay() -> dict[str, Any]: """Форма build_forecast_overlay (#983) — класс §10.2 / ranked_segments / USP / commercial.""" return { "horizon_months": 12, "mode": "demand_supply", "advisory": True, "ranked_segments": [ { "bucket": "1-Студия", "obj_class": "комфорт", "deficit_index": 0.34, "confidence": "medium", }, { "bucket": "2-1-к", "obj_class": "комфорт", "deficit_index": 0.22, "confidence": "medium", }, ], "warnings": [], "class_reco": { "obj_class": "комфорт", "mean_deficit_index": 0.28, "n_segments": 2, "reason": { "why": "Класс «комфорт»: сильнейший средний дефицит.", "drivers": [{"factor": "deficit_index", "value": 0.28, "direction": "+"}], "rejected": [], "what_would_change": ["Рост ставки → спрос мягче."], "confidence": "medium", "advisory": True, }, }, "usp": [ { "segment": "1-Студия", "obj_class": "комфорт", "deficit_index": 0.34, "usp_text": "Дефицит формата «1-Студия (комфорт)» — стройте его.", } ], "commercial": {"available": False, "caveat": "коммерция: нет достаточных данных"}, } def _sample_product_scores() -> dict[str, Any]: """Форма ProductScoreCard.as_dict() (#985, 10 скоров + overall).""" return { "segment": {"obj_class": "комфорт", "room_bucket": "1-к 30-45"}, "horizon_months": 12, "scores": { "market_fit": {"key": "market_fit", "value": 0.67, "confidence": "medium"}, "demand": {"key": "demand", "value": 0.51, "confidence": "medium"}, }, "overall": 0.62, "advisory": True, "confidence": "medium", } def _sample_special_indices() -> dict[str, Any]: """Форма SpecialIndices.as_dict() (#986, 6 индексов).""" return { "segment": {"obj_class": "комфорт"}, "district": "Верх-Исетский", "indices": { "launch_window": {"key": "launch_window", "value": 0.6, "label": "12 мес"}, "product_void": {"key": "product_void", "value": 0.4, "label": "2 белых пятна"}, }, "advisory": True, "confidence": "medium", } def _full_assemble() -> SiteFinderReport: """Полный сбор отчёта из sample analyze dict + sample advisory dict'ов всех секций.""" return assemble_report( _sample_analyze(), market_metrics=_sample_market_metrics(), supply_layers=_sample_supply_layers(), forecasts=_sample_forecasts(), future_supply=_sample_future_supply(), scenarios=_sample_scenarios(), recommendation_overlay=_sample_overlay(), product_scores=_sample_product_scores(), special_indices=_sample_special_indices(), segment={"obj_class": "комфорт", "room_bucket": "1-к 30-45"}, cad_num="66:41:0000000:1", district="Верх-Исетский", ) # ── Полный сбор: восемь секций + JSON-serializability + advisory ────────────── class TestFullAssemble: def test_returns_site_finder_report(self) -> None: assert isinstance(_full_assemble(), SiteFinderReport) def test_all_eight_sections_present_and_dicts(self) -> None: payload = _full_assemble().as_dict() for key in _SECTION_KEYS: assert key in payload, f"отсутствует секция {key}" assert isinstance(payload[key], dict) def test_as_dict_is_json_serializable(self) -> None: # Главный контракт #987: as_dict() проходит json.dumps без default= (ничего # сырого/dataclass'ов в выдаче). Полный round-trip (json.loads == payload) на # частичном/пустом отчёте ниже — здесь scenarios.rate_path несёт int-ключи # (как ScenarioForecast.as_dict()), которые JSON приводит к строкам, поэтому # строгое равенство неинформативно — важна именно сериализуемость. payload = _full_assemble().as_dict() dumped = json.dumps(payload, ensure_ascii=False) assert isinstance(dumped, str) assert json.loads(dumped) is not None def test_advisory_true_everywhere(self) -> None: payload = _full_assemble().as_dict() assert payload["advisory"] is True assert payload["meta"]["generated_advisory"] is True def test_schema_version_present(self) -> None: payload = _full_assemble().as_dict() assert payload["schema_version"] == payload["meta"]["schema_version"] # ── market_now ← analyze + §9.2/§9.3 ────────────────────────────────────────── class TestMarketNow: def test_market_metrics_and_supply_passed_through(self) -> None: market_now = _full_assemble().as_dict()["market_now"] assert market_now["market_metrics"]["unit_velocity"] == 8.2 assert market_now["supply_layers"]["open_units"] == 1200 def test_competitors_from_analyze(self) -> None: market_now = _full_assemble().as_dict()["market_now"] assert len(market_now["competitors"]) == 2 assert market_now["competitors"][0]["comm_name"] == "ЖК Альфа" def test_summary_synthesized(self) -> None: market_now = _full_assemble().as_dict()["market_now"] assert market_now["summary"] is not None assert "абсорбция" in market_now["summary"] # ── future_market ← forecasts / future_supply / scenarios ───────────────────── class TestFutureMarket: def test_forecasts_by_horizon(self) -> None: fm = _full_assemble().as_dict()["future_market"] assert len(fm["forecasts_by_horizon"]) == 4 assert fm["forecasts_by_horizon"][1]["deficit_index"] == 0.34 def test_months_of_inventory_flows_through(self) -> None: # MOI прокидывается per-горизонт автоматически (passthrough as_dict()). fm = _full_assemble().as_dict()["future_market"] assert fm["forecasts_by_horizon"][1]["months_of_inventory"] == 24.2 def test_summary_mentions_months_of_inventory(self) -> None: # future_market.summary несёт ДИСКРИМИНИРУЮЩИЙ MOI-headline рядом с дефицитом. fm = _full_assemble().as_dict()["future_market"] assert fm["summary"] is not None assert "мес конкурирующего предложения" in fm["summary"] def test_future_supply_passed_through(self) -> None: fm = _full_assemble().as_dict()["future_market"] assert fm["future_supply"]["index"] == 0.42 def test_future_competitors_from_primary_horizon(self) -> None: fm = _full_assemble().as_dict()["future_market"] # Будущие конкуренты подняты с горизонта 12 мес (там список непуст). assert len(fm["future_competitors"]) == 1 assert fm["future_competitors"][0]["comm_name"] == "ЖК Гамма" def test_scenarios_summary_deficit_spread(self) -> None: fm = _full_assemble().as_dict()["future_market"] assert fm["scenarios_summary"]["conservative"] == 0.18 assert fm["scenarios_summary"]["aggressive"] == 0.49 # ── product_tz ← overlay #983 ───────────────────────────────────────────────── class TestProductTz: def test_class_from_overlay(self) -> None: pt = _full_assemble().as_dict()["product_tz"] assert pt["obj_class"] == "комфорт" def test_mix_from_ranked_segments(self) -> None: pt = _full_assemble().as_dict()["product_tz"] assert len(pt["mix"]) == 2 assert pt["mix"][0]["bucket"] == "1-Студия" def test_usp_and_commercial(self) -> None: pt = _full_assemble().as_dict()["product_tz"] assert len(pt["usp"]) == 1 assert pt["commercial"]["available"] is False def test_reasons_lifted_from_class_reco(self) -> None: pt = _full_assemble().as_dict()["product_tz"] assert len(pt["reasons"]) == 1 assert "комфорт" in pt["reasons"][0]["why"] # ── scenarios ← #984 by_scenario ────────────────────────────────────────────── class TestScenarios: def test_by_scenario_keyed(self) -> None: sc = _full_assemble().as_dict()["scenarios"] assert set(sc["by_scenario"].keys()) == {"conservative", "base", "aggressive"} def test_summary_present(self) -> None: sc = _full_assemble().as_dict()["scenarios"] assert sc["summary"] is not None # ── scoring ← #985 + #986 ───────────────────────────────────────────────────── class TestScoring: def test_product_scores_and_overall(self) -> None: scoring = _full_assemble().as_dict()["scoring"] assert scoring["product_scores"]["overall"] == 0.62 assert scoring["overall"] == 0.62 def test_special_indices_passed_through(self) -> None: scoring = _full_assemble().as_dict()["scoring"] assert scoring["special_indices"]["indices"]["launch_window"]["value"] == 0.6 # ── confidence ← compute_report_confidence (#990) ───────────────────────────── class TestConfidence: def test_level_computed(self) -> None: conf = _full_assemble().as_dict()["confidence"] assert conf["level"] in ("high", "medium", "low") def test_advisory_never_high(self) -> None: # advisory=True → #990 cap 'medium' (НИКОГДА 'high'). conf = _full_assemble().as_dict()["confidence"] assert conf["level"] != "high" def test_structural_rationale_present(self) -> None: conf = _full_assemble().as_dict()["confidence"] assert isinstance(conf["rationale"], str) and conf["rationale"] def test_factors_carry_advisory_capped(self) -> None: conf = _full_assemble().as_dict()["confidence"] assert "advisory_capped" in conf["factors"] def test_data_quality_factors_extracted(self) -> None: # deal_count (n_sold=88) / analog_count (obj_count=7) / domrf_coverage (0.55) # извлечены и попали в факторы #990. conf = _full_assemble().as_dict()["confidence"] factors = conf["factors"] assert "deal_count" in factors assert "analog_count" in factors assert "domrf_coverage" in factors def test_confounded_window_factor_surfaces_when_any_forecast_confounded(self) -> None: # #1222 end-to-end: хотя бы один forecast.confounded=True → confounded_window # фактор присутствует и тянет уровень в 'low' (#990: confounded fact-level low, # weakest-link → итог 'low'; advisory cap ниже 'low' не двигает). forecasts = _sample_forecasts() forecasts[2]["confounded"] = True # 18-мес forecast confounded report = assemble_report( _sample_analyze(), market_metrics=_sample_market_metrics(), supply_layers=_sample_supply_layers(), forecasts=forecasts, future_supply=_sample_future_supply(), scenarios=_sample_scenarios(), recommendation_overlay=_sample_overlay(), product_scores=_sample_product_scores(), special_indices=_sample_special_indices(), cad_num="66:41:0000000:1", district="Верх-Исетский", ).as_dict() factors = report["confidence"]["factors"] assert "confounded_window" in factors, "шок-окно перманентно мёртв — #1222 регрессия" assert factors["confounded_window"]["value"] is True assert factors["confounded_window"]["level"] == "low" # Weakest-link MIN → итоговый уровень тоже 'low' (был бы 'medium' без шока). assert report["confidence"]["level"] == "low" def test_confounded_window_factor_absent_when_no_forecast_confounded(self) -> None: # Зеркальный кейс: все forecast'ы confounded=False → factor отсутствует # (confidence_engine добавляет confounded factor ТОЛЬКО при True, line 450 — # чистое окно не тянет искусственно вверх, иначе тонкие отчёты получили бы # фантомный 'high'-вклад). Гарантия, что previous test НЕ false-positive # (factor не «всегда low»), достигается тем, что предыдущий тест проверяет # уровень: «low» доступен только если confounded factor реально добавлен. conf = _full_assemble().as_dict()["confidence"] factors = conf["factors"] assert "confounded_window" not in factors # ── _component_confidences — per-horizon forecast collapse (#1958) ──────────── class TestComponentConfidencesForecastCollapse: """#1958: forecasts отдаёт confidence НА КАЖДЫЙ горизонт; раньше каждый горизонт эмитился отдельной парой → «Прогноз спрос/предложение» дублировался ×N в факторах. Теперь сворачиваем в ОДНУ пару weakest-link'ом (MIN ранга по горизонтам).""" def test_four_horizons_collapse_to_one_forecast_pair(self) -> None: forecasts = [ {"horizon_months": 6, "confidence": "high"}, {"horizon_months": 12, "confidence": "high"}, {"horizon_months": 18, "confidence": "high"}, {"horizon_months": 24, "confidence": "high"}, ] out = _component_confidences(None, None, forecasts, None, None) forecast_pairs = [p for p in out if p[0] == "forecasts"] assert len(forecast_pairs) == 1, "per-горизонт forecast confidences не свёрнуты (#1958)" assert forecast_pairs[0] == ("forecasts", "high") def test_weakest_link_min_across_horizons(self) -> None: # Худший горизонт (low) тянет общий forecast-фактор вниз. forecasts = [ {"horizon_months": 6, "confidence": "high"}, {"horizon_months": 12, "confidence": "medium"}, {"horizon_months": 18, "confidence": "low"}, {"horizon_months": 24, "confidence": "high"}, ] out = _component_confidences(None, None, forecasts, None, None) forecast_pairs = [p for p in out if p[0] == "forecasts"] assert forecast_pairs == [("forecasts", "low")] def test_no_forecast_pair_when_all_confidences_missing(self) -> None: forecasts = [ {"horizon_months": 6}, # нет confidence {"horizon_months": 12, "confidence": "garbage"}, # не whitelisted ] out = _component_confidences(None, None, forecasts, None, None) assert [p for p in out if p[0] == "forecasts"] == [] def test_other_service_pairs_intact(self) -> None: # Свёртка forecasts не должна задевать остальные per-service факторы. out = _component_confidences( {"confidence": "medium"}, # market_metrics {"confidence": "high"}, # future_supply [{"horizon_months": 12, "confidence": "low"}], {"confidence": "high"}, # product_scores {"confidence": "medium"}, # special_indices ) names = [p[0] for p in out] assert names.count("forecasts") == 1 assert ("market_metrics", "medium") in out assert ("future_supply", "high") in out assert ("product_scores", "high") in out assert ("special_indices", "medium") in out # ── exec_summary — синтез ───────────────────────────────────────────────────── class TestExecSummary: def test_headline_non_empty(self) -> None: es = _full_assemble().as_dict()["exec_summary"] assert isinstance(es["headline"], str) and es["headline"] def test_headline_mentions_class(self) -> None: es = _full_assemble().as_dict()["exec_summary"] assert "комфорт" in es["headline"] def test_verdict_and_key_numbers(self) -> None: es = _full_assemble().as_dict()["exec_summary"] assert es["verdict"] is not None assert es["key_numbers"]["deficit_index"] == 0.34 assert es["key_numbers"]["months_of_inventory"] == 24.2 assert es["key_numbers"]["overall_score"] == 0.62 def test_verdict_mentions_months_of_inventory(self) -> None: # ДИСКРИМИНИРУЮЩИЙ MOI-headline присутствует в вердикте exec_summary. es = _full_assemble().as_dict()["exec_summary"] assert "мес конкурирующего предложения" in es["verdict"] def test_overall_confidence_matches_section(self) -> None: payload = _full_assemble().as_dict() assert payload["exec_summary"]["overall_confidence"] == payload["confidence"]["level"] # ── meta ────────────────────────────────────────────────────────────────────── class TestMeta: def test_context_fields(self) -> None: meta = _full_assemble().as_dict()["meta"] assert meta["cad_num"] == "66:41:0000000:1" assert meta["district"] == "Верх-Исетский" assert meta["segment"]["obj_class"] == "комфорт" assert meta["horizons"] == [6, 12, 18, 24] # ── Частичный сбор: ТОЛЬКО analyze dict → валидный частичный отчёт ───────────── class TestPartialAssemble: def _partial(self) -> SiteFinderReport: return assemble_report(_sample_analyze(), cad_num="66:41:0000000:1") def test_valid_partial_report(self) -> None: report = self._partial() assert isinstance(report, SiteFinderReport) payload = report.as_dict() # Все восемь секций присутствуют (контракт стабилен), JSON-safe. for key in _SECTION_KEYS: assert key in payload assert json.loads(json.dumps(payload, ensure_ascii=False)) == payload def test_market_now_filled_from_analyze(self) -> None: payload = self._partial().as_dict() # competitors из analyze есть; market_metrics/supply_layers None (не переданы). assert len(payload["market_now"]["competitors"]) == 2 assert payload["market_now"]["market_metrics"] is None assert payload["market_now"]["supply_layers"] is None def test_future_sections_empty(self) -> None: payload = self._partial().as_dict() assert payload["future_market"]["forecasts_by_horizon"] == [] assert payload["future_market"]["future_supply"] is None assert payload["scenarios"]["by_scenario"] == {} assert payload["scoring"]["product_scores"] is None assert payload["product_tz"]["obj_class"] is None def test_confidence_capped_advisory_on_partial(self) -> None: # Без forecasts/metrics #990 видит из analyze только analog_count (12 → # high) + domrf_coverage (75% → 0.75 → high); None-сигналы (deal_count/ # history) #990 в факторы НЕ добавляет (None ≠ low-фактор). Weakest-link → # high, но advisory-cap опускает до 'medium' (НИКОГДА не 'high'). payload = self._partial().as_dict() assert payload["confidence"]["level"] == "medium" assert payload["confidence"]["level"] != "high" assert payload["confidence"]["rationale"] def test_exec_summary_still_has_headline(self) -> None: payload = self._partial().as_dict() assert payload["exec_summary"]["headline"] def test_advisory_true(self) -> None: assert self._partial().as_dict()["advisory"] is True # ── Graceful: пустой analyze {} → отчёт всё равно валиден ────────────────────── class TestGracefulEmpty: def test_empty_analyze_valid_report(self) -> None: report = assemble_report({}) payload = report.as_dict() for key in _SECTION_KEYS: assert key in payload assert json.loads(json.dumps(payload, ensure_ascii=False)) == payload assert payload["advisory"] is True # Всё тонко → confidence 'low', headline честный «недостаточно данных». assert payload["confidence"]["level"] == "low" assert payload["exec_summary"]["headline"] def test_all_none_inputs(self) -> None: # Явные None всех под-сервисов — не должно бросать, отчёт валиден. report = assemble_report( {}, market_metrics=None, supply_layers=None, forecasts=None, future_supply=None, scenarios=None, recommendation_overlay=None, product_scores=None, special_indices=None, ) assert isinstance(report, SiteFinderReport) # ── Вход КАК dataclass, ТАК и as_dict-словарь → нормализуется ────────────────── @dataclass(frozen=True) class _FakeMetrics: """Минимальный stand-in под-сервиса с `as_dict()` (как реальный MarketMetrics).""" unit_velocity: float obj_count: int n_sold: int def as_dict(self) -> dict[str, Any]: return { "unit_velocity": self.unit_velocity, "obj_count": self.obj_count, "n_sold": self.n_sold, "confidence": "medium", "window_months": 18, } class TestInputNormalization: def test_dataclass_instance_accepted(self) -> None: # Передаём ОБЪЕКТ (не as_dict) — сборщик нормализует через _as_dict_or. report = assemble_report(_sample_analyze(), market_metrics=_FakeMetrics(8.2, 7, 88)) market_now = report.as_dict()["market_now"] assert market_now["market_metrics"]["unit_velocity"] == 8.2 def test_dataclass_and_dict_give_same_market_metrics(self) -> None: from_obj = assemble_report( _sample_analyze(), market_metrics=_FakeMetrics(8.2, 7, 88) ).as_dict() from_dict = assemble_report( _sample_analyze(), market_metrics=_FakeMetrics(8.2, 7, 88).as_dict() ).as_dict() assert from_obj["market_now"]["market_metrics"] == from_dict["market_now"]["market_metrics"] # ── Pure-хелперы извлечения сигналов (юнит, без БД) ──────────────────────────── class TestSignalExtractionHelpers: def test_as_dict_or_normalizes(self) -> None: assert _as_dict_or(None) is None assert _as_dict_or({"a": 1}) == {"a": 1} assert _as_dict_or(_FakeMetrics(1.0, 2, 3))["obj_count"] == 2 assert _as_dict_or(42) is None # мусор → None (graceful) def test_deal_count_from_market_metrics(self) -> None: assert _deal_count({}, {"n_sold": 88}) == 88 assert _deal_count({}, None) is None def test_analog_count_prefers_obj_count(self) -> None: assert _analog_count({"market_pulse": {"competitors_total": 12}}, {"obj_count": 7}) == 7 def test_analog_count_fallback_to_analyze(self) -> None: assert _analog_count({"market_pulse": {"competitors_total": 12}}, None) == 12 assert _analog_count({"competitors": [{}, {}, {}]}, None) == 3 assert _analog_count({}, None) is None def test_domrf_coverage_fraction_and_percent(self) -> None: # supply_layers доля (0.55) → как есть; analyze проценты (75.0) → /100. assert _domrf_coverage({}, {"domrf_coverage": 0.55}) == 0.55 assert _domrf_coverage({"market_data_coverage_pct": 75.0}, None) == 0.75 assert _domrf_coverage({}, None) is None def test_domrf_coverage_sub_one_percent_not_read_as_fraction(self) -> None: # BUG #3: настоящий sub-1% процент (0.8% покрытия) — percent-ветка делит на 100 # → 0.008, НЕ 0.8 (что было бы 80% и инфлировало бы confidence в exactly # near-zero кейсе, который §15 призван флагать). assert _domrf_coverage({"market_data_coverage_pct": 0.8}, None) == 0.008 # Каноничный sparse-сигнал проекта ~2.5%. assert _domrf_coverage({"market_data_coverage_pct": 2.5}, None) == 0.025 def test_domrf_coverage_fraction_source_unchanged(self) -> None: # supply_layers — уже доля: 0.025 остаётся 0.025 (НЕ делим на 100). assert _domrf_coverage({}, {"domrf_coverage": 0.025}) == 0.025 def test_history_months_from_window(self) -> None: assert _history_months({"window_months": 18}, []) == 18 assert _history_months(None, []) is None def test_confounded_any_horizon(self) -> None: assert _confounded([{"confounded": False}, {"confounded": True}]) is True assert _confounded([{"is_confounded_window": True}]) is True assert _confounded([{"confounded": False}]) is False assert _confounded([]) is False def test_confounded_missing_key_is_false_not_keyerror(self) -> None: # #1222: forecast БЕЗ confounded-ключа НЕ должен ронять (.get() default → None # ≠ True → False). Раньше .get() уже стоял, но это контракт-тест: гарантирует, # что defensive-чтение НЕ деградирует в KeyError при произвольных формах. assert _confounded([{}, {"horizon_months": 12}]) is False assert _confounded([{"confounded": "no"}]) is False # truthy-but-not-True → False assert _confounded([{"confounded": 1}]) is False # «is True» строго, не truthy def test_confounded_reads_real_demand_supply_forecast_as_dict(self) -> None: # #1222 regression: DemandSupplyForecast.as_dict() ОБЯЗАН нести ключ `confounded`, # иначе шок-окно §15 (#990) перманентно мёртв. Строим РЕАЛЬНЫЙ frozen-dataclass # БЕЗ БД, сериализуем через его собственный `as_dict()` и убеждаемся, что # `_confounded` видит флаг (контракт-тест: ловит дрейф ключа продьюсера). from app.services.forecasting.demand_supply_forecast import DemandSupplyForecast clean = 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=100.0, open_units=300, hidden_release_units=80.0, future_online_units=20.0, projected_supply_units=400.0, balance_units=-300.0, balance_ratio=0.25, deficit_index=-0.5, months_of_inventory=48.0, rate_future=18.0, rate_sensitivity_phrase=None, future_competitors=[], advisory=True, confidence="medium", confounded=False, ).as_dict() shock = DemandSupplyForecast( segment={"obj_class": "комфорт"}, horizon_months=24, base_pace_units_per_mo=8.0, demand_norm_coefficient=1.0, macro_coefficient=1.0, projected_demand_units=200.0, open_units=300, hidden_release_units=80.0, future_online_units=20.0, projected_supply_units=400.0, balance_units=-200.0, balance_ratio=0.5, deficit_index=-0.3, months_of_inventory=24.0, rate_future=18.0, rate_sensitivity_phrase=None, future_competitors=[], advisory=True, confidence="medium", confounded=True, ).as_dict() assert "confounded" in clean and "confounded" in shock # контракт ключа assert _confounded([clean]) is False assert _confounded([clean, shock]) is True def test_primary_deficit_prefers_12mo(self) -> None: forecasts = [ {"horizon_months": 6, "deficit_index": 0.21}, {"horizon_months": 12, "deficit_index": 0.34}, ] assert _primary_deficit_index(forecasts) == 0.34 def test_primary_deficit_fallback_first(self) -> None: forecasts = [{"horizon_months": 6, "deficit_index": 0.21}] assert _primary_deficit_index(forecasts) == 0.21 assert _primary_deficit_index([]) is None def test_primary_moi_prefers_12mo(self) -> None: forecasts = [ {"horizon_months": 6, "months_of_inventory": 18.5}, {"horizon_months": 12, "months_of_inventory": 24.2}, ] assert _primary_months_of_inventory(forecasts) == 24.2 def test_primary_moi_fallback_first(self) -> None: forecasts = [{"horizon_months": 6, "months_of_inventory": 18.5}] assert _primary_months_of_inventory(forecasts) == 18.5 assert _primary_months_of_inventory([]) is None def test_primary_moi_none_when_absent(self) -> None: # forecast без MOI-ключа → None (graceful, не KeyError). assert _primary_months_of_inventory([{"horizon_months": 12, "deficit_index": -1.0}]) is None