feat(forecast): forecast Celery task + persist §22 «1.0» (Step 3b-i) #1068
6 changed files with 599 additions and 6 deletions
|
|
@ -178,18 +178,46 @@ def latest_run_dates(db: Session, cad_nums: list[str]) -> dict[str, datetime]:
|
|||
return {row["cad_num"]: row["created_at"] for row in rows}
|
||||
|
||||
|
||||
def latest_run_for(db: Session, cad_num: str) -> Row[Any] | None:
|
||||
"""Последний ран на участок целиком (для re-open / «текущий анализ участка»).
|
||||
def latest_run_for(
|
||||
db: Session,
|
||||
cad_num: str,
|
||||
*,
|
||||
schema_version: str | None = None,
|
||||
) -> Row[Any] | None:
|
||||
"""Последний ран на участок (для re-open / база форсайта по конкретной схеме).
|
||||
|
||||
Читает из v_analysis_runs_latest (DISTINCT ON cad_num, max created_at). None если
|
||||
участок ещё не анализировался. Возвращает Row (id/result/params/.../created_at).
|
||||
Два режима:
|
||||
• schema_version=None (дефолт, backward-compatible): читает из
|
||||
v_analysis_runs_latest (DISTINCT ON cad_num, max created_at) — последний ран
|
||||
ЛЮБОЙ схемы. Это исходное поведение, существующие вызовы не меняются.
|
||||
• schema_version задан: читает из БАЗОВОЙ таблицы analysis_runs с фильтром по
|
||||
schema_version — последний ран ИМЕННО этой схемы. Нужно форсайт-таске (3b-i),
|
||||
чтобы достать последний site-analysis ран ('analyze-1.0') как базу прогноза,
|
||||
даже когда поверх него уже легли §22-раны ('1.0') — view бы вернул §22-ран.
|
||||
|
||||
None если подходящего рана нет. Возвращает Row (id/result/params/.../created_at).
|
||||
psycopg v3: CAST(:x AS text) — никогда :x::type (backend.md).
|
||||
"""
|
||||
if schema_version is None:
|
||||
return db.execute(
|
||||
text("""
|
||||
SELECT id, cad_num, district, segment, params, result,
|
||||
schema_version, advisory, confidence, status, created_by,
|
||||
created_at
|
||||
FROM v_analysis_runs_latest
|
||||
WHERE cad_num = CAST(:cad_num AS text)
|
||||
"""),
|
||||
{"cad_num": cad_num},
|
||||
).first()
|
||||
return db.execute(
|
||||
text("""
|
||||
SELECT id, cad_num, district, segment, params, result,
|
||||
schema_version, advisory, confidence, status, created_by, created_at
|
||||
FROM v_analysis_runs_latest
|
||||
FROM analysis_runs
|
||||
WHERE cad_num = CAST(:cad_num AS text)
|
||||
AND schema_version = CAST(:schema_version AS text)
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
"""),
|
||||
{"cad_num": cad_num},
|
||||
{"cad_num": cad_num, "schema_version": schema_version},
|
||||
).first()
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ celery_app = Celery(
|
|||
"app.workers.tasks.ekburg_permits_sync",
|
||||
"app.workers.tasks.cbr_macro_sync",
|
||||
"app.workers.tasks.supply_layers_refresh",
|
||||
"app.workers.tasks.forecast",
|
||||
],
|
||||
)
|
||||
celery_app.conf.timezone = "Europe/Moscow"
|
||||
|
|
|
|||
174
backend/app/workers/tasks/forecast.py
Normal file
174
backend/app/workers/tasks/forecast.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
"""Celery task: посчитать §22 SiteFinderReport в фоне и сохранить в analysis_runs.
|
||||
|
||||
Step 3b-i проводки форсайта. §22-оркестратор ``build_site_finder_report`` (#1056,
|
||||
forecasting/orchestrator.py) считает форсайт из §9.x-слоёв — это занимает ~30-180с
|
||||
(ЕКБ-wide ~184с, scoped-сегмент меньше), поэтому НЕ может крутиться inline на
|
||||
синхронном /analyze-эндпоинте. Здесь — фоновая Celery-таска, которая его запускает
|
||||
и персистит результат рядом с site-analysis раном в той же таблице ``analysis_runs``.
|
||||
|
||||
ПОТОК (read → build → persist):
|
||||
1. Читаем ПОСЛЕДНИЙ site-analysis ран ('analyze-1.0') для cad_num через
|
||||
schema-фильтрованный ``latest_run_for(..., schema_version=ANALYZE_SCHEMA_VERSION)``
|
||||
(#3b-i, repository.py) — НЕ через view: поверх analyze-рана уже могут лежать
|
||||
§22-раны ('1.0'), view вернул бы их, а нам нужна именно site-analysis база.
|
||||
Нет такого рана → logger.warning + return None (нечего форсайтить — graceful,
|
||||
без падения воркера).
|
||||
2. ``analyze = row.result`` (полный персиснутый dict analyze — несёт district +
|
||||
competitors, которые нужны оркестратору). district — из денорм-колонки рана,
|
||||
fallback в analyze["district"]["district_name"].
|
||||
3. ``build_site_finder_report(db, analyze=…, cad_num=…, district=…, horizons=…)``
|
||||
— горизонты = §9.x default grid, с гарантией что запрошенный ``horizon`` в сетке.
|
||||
4. ``report.as_dict()`` → ``persist_analysis_run(... schema_version="1.0",
|
||||
status="done" ...)``. persist сам best-effort (SAVEPOINT + commit), нормализует
|
||||
status/confidence под CHECK'и миграции 127 (status='done' → нормализуется в
|
||||
'complete'; форма CHECK не знает 'done'). Возвращаем id новой строки (или None).
|
||||
|
||||
NB схема: site-analysis раны несут schema_version='analyze-1.0' (ANALYZE_SCHEMA_VERSION),
|
||||
§22-форсайт-раны — '1.0' (собственный SiteFinderReport._SCHEMA_VERSION, прокидывается
|
||||
из ``d["schema_version"]``). Оба сосуществуют в analysis_runs, различаются по
|
||||
schema_version — поэтому read берёт именно 'analyze-1.0', а write кладёт '1.0'.
|
||||
|
||||
Mirror conventions (supply_layers_refresh.py / cbr_macro_sync.py): ``@celery_app.task``,
|
||||
``SessionLocal()`` + try/finally close, ``logger`` (не print), синхронный ``def`` (НЕ
|
||||
async внутри таски), ``CAST(:x AS type)`` живёт в repository (тут SQL не пишем).
|
||||
|
||||
TIME LIMIT: build_site_finder_report может занять минуты — ставим щедрый per-task лимит
|
||||
(time_limit=900с / soft_time_limit=840с), чтобы таску не убили mid-compute. Глобального
|
||||
task_time_limit в celery_app.py нет (только timezone + beat_schedule), так что override
|
||||
тут — единственный лимит для этой таски.
|
||||
|
||||
REGISTRATION: добавлено в ``include=[...]`` celery_app.py (зеркало того, как
|
||||
зарегистрирован supply_layers_refresh) — beat-расписания НЕТ намеренно (таска
|
||||
enqueue-on-demand; триггер из /analyze — отдельный шаг 3b-ii, вне scope).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
from app.services.analysis_runs.repository import (
|
||||
ANALYZE_SCHEMA_VERSION,
|
||||
latest_run_for,
|
||||
persist_analysis_run,
|
||||
)
|
||||
from app.services.forecasting.orchestrator import build_site_finder_report
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Горизонты прогноза по умолчанию (мес) — зеркало §9.x default grid (orchestrator.py
|
||||
# _DEFAULT_HORIZONS / #952/#984/#986). Запрошенный horizon домешивается, если его нет.
|
||||
_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)
|
||||
|
||||
|
||||
def _horizons_for(horizon: int) -> list[int]:
|
||||
"""Default-сетка горизонтов с ГАРАНТИЕЙ что запрошенный ``horizon`` в ней. PURE.
|
||||
|
||||
§9.x single-горизонтные сервисы (score_card/overlay/future_supply) берут primary
|
||||
из сетки (orchestrator._primary_horizon → 12 если есть, иначе первый). Чтобы
|
||||
запрошенный horizon реально влиял на прогноз, добавляем его в сетку (отсортировано,
|
||||
без дублей). horizon ≤ 0 (бессмысленный) → возвращаем чистую default-сетку.
|
||||
"""
|
||||
if horizon <= 0:
|
||||
return list(_DEFAULT_HORIZONS)
|
||||
return sorted(set(_DEFAULT_HORIZONS) | {horizon})
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
bind=True,
|
||||
name="tasks.forecast.forecast_site_finder_report",
|
||||
time_limit=900,
|
||||
soft_time_limit=840,
|
||||
)
|
||||
def forecast_site_finder_report(
|
||||
self: Any,
|
||||
cad_num: str,
|
||||
horizon: int = 12,
|
||||
created_by: str | None = None,
|
||||
) -> int | None:
|
||||
"""Посчитать §22 SiteFinderReport для участка в фоне и сохранить в analysis_runs.
|
||||
|
||||
Берёт последний site-analysis ран ('analyze-1.0') как базу, считает §22-форсайт
|
||||
через оркестратор (#1056) и персистит ``SiteFinderReport.as_dict()`` отдельной
|
||||
строкой ('1.0') в той же таблице. Best-effort на каждом шаге: нет базы → None
|
||||
(warning, не crash); неожиданная ошибка compute → logger.exception + None (таска
|
||||
не валит воркер — persist сам best-effort, и форсайт может честно не собраться).
|
||||
|
||||
Args:
|
||||
cad_num: кадастровый номер участка.
|
||||
horizon: основной горизонт прогноза (мес); домешивается в default-сетку.
|
||||
created_by: автор рана (прокидывается в persist; None допустимо).
|
||||
|
||||
Returns:
|
||||
id новой строки analysis_runs (§22-форсайт, schema_version='1.0') или None,
|
||||
если базы для прогноза нет / персист не удался / compute упал.
|
||||
"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# 1. База прогноза — последний site-analysis ран ('analyze-1.0'), НЕ через view.
|
||||
row = latest_run_for(db, cad_num, schema_version=ANALYZE_SCHEMA_VERSION)
|
||||
if row is None:
|
||||
logger.warning(
|
||||
"forecast_site_finder_report: no %s run for cad=%s — nothing to forecast "
|
||||
"from (run /analyze first); returning None",
|
||||
ANALYZE_SCHEMA_VERSION,
|
||||
cad_num,
|
||||
)
|
||||
return None
|
||||
|
||||
# 2. analyze-dict (несёт district + competitors) + district (денорм → fallback).
|
||||
analyze: dict[str, Any] = row.result or {}
|
||||
district = row.district or (analyze.get("district") or {}).get("district_name")
|
||||
|
||||
horizons = _horizons_for(horizon)
|
||||
logger.info(
|
||||
"forecast_site_finder_report: building §22 report cad=%s district=%s "
|
||||
"horizon=%s horizons=%s (base run id=%s)",
|
||||
cad_num,
|
||||
district,
|
||||
horizon,
|
||||
horizons,
|
||||
row.id,
|
||||
)
|
||||
|
||||
# 3. §22-форсайт через оркестратор (#1056) — может занять минуты (time_limit=900).
|
||||
report = build_site_finder_report(
|
||||
db,
|
||||
analyze=analyze,
|
||||
cad_num=cad_num,
|
||||
district=district,
|
||||
horizons=horizons,
|
||||
)
|
||||
d = report.as_dict()
|
||||
|
||||
# 4. Персист отдельной строкой '1.0'. status='done' нормализуется repository'ем
|
||||
# под CHECK миграции 127 (→ 'complete'); confidence/segment — из as_dict().
|
||||
new_id = persist_analysis_run(
|
||||
db,
|
||||
cad_num=cad_num,
|
||||
result=d,
|
||||
params={"horizon": horizon, "source": "forecast-task"},
|
||||
district=district,
|
||||
confidence=(d.get("confidence") or {}).get("level"),
|
||||
status="done",
|
||||
schema_version=d["schema_version"], # "1.0" (SiteFinderReport._SCHEMA_VERSION)
|
||||
created_by=created_by,
|
||||
segment=(d.get("meta") or {}).get("segment"),
|
||||
)
|
||||
logger.info(
|
||||
"forecast_site_finder_report: persisted §22 report cad=%s run_id=%s schema_version=%s",
|
||||
cad_num,
|
||||
new_id,
|
||||
d.get("schema_version"),
|
||||
)
|
||||
return new_id
|
||||
except Exception:
|
||||
# Неожиданный сбой compute/чтения: логируем traceback и НЕ роняем воркер —
|
||||
# форсайт может честно не собраться (тонкие данные / сбой §9.x-шва), а persist
|
||||
# сам best-effort. Возвращаем None — таска failed gracefully.
|
||||
logger.exception("forecast_site_finder_report failed for cad=%s (returning None)", cad_num)
|
||||
return None
|
||||
finally:
|
||||
db.close()
|
||||
0
backend/tests/services/analysis_runs/__init__.py
Normal file
0
backend/tests/services/analysis_runs/__init__.py
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
"""Тесты schema-фильтрованного чтения latest_run_for (#3b-i, repository.py).
|
||||
|
||||
latest_run_for получил keyword-only ``schema_version``:
|
||||
• None (дефолт): читает из v_analysis_runs_latest (последний ран ЛЮБОЙ схемы) —
|
||||
backward-compatible, существующие вызовы не меняются.
|
||||
• задан: читает из БАЗОВОЙ таблицы analysis_runs с фильтром schema_version —
|
||||
последний ран ИМЕННО этой схемы (нужно форсайт-таске достать 'analyze-1.0' базу
|
||||
даже когда поверх легли §22-раны '1.0', которые вернул бы view).
|
||||
|
||||
Чистые тесты (mock session, без живой БД): проверяем форму SQL (какая таблица /
|
||||
фильтр / ORDER BY) и что фильтрованный режим прокидывает schema_version в params и
|
||||
возвращает именно отфильтрованную строку. psycopg v3: CAST(:x AS text), не :x::type.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from app.services.analysis_runs.repository import ANALYZE_SCHEMA_VERSION, latest_run_for
|
||||
|
||||
|
||||
def _mock_db_capturing() -> tuple[MagicMock, list[tuple[Any, Any]]]:
|
||||
"""Mock session: захватывает (stmt, params) каждого execute, .first() → sentinel-row.
|
||||
|
||||
Возвращает (db, captured) — captured[i] = (stmt, params) i-го execute. Сам Row
|
||||
мокаем как MagicMock (тесту достаточно, что вернулся именно он).
|
||||
"""
|
||||
captured: list[tuple[Any, Any]] = []
|
||||
sentinel_row = MagicMock(name="row")
|
||||
|
||||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||||
captured.append((stmt, params))
|
||||
result = MagicMock()
|
||||
result.first.return_value = sentinel_row
|
||||
return result
|
||||
|
||||
db = MagicMock()
|
||||
db.execute = execute_router
|
||||
db._sentinel_row = sentinel_row # для assert identity
|
||||
return db, captured
|
||||
|
||||
|
||||
def test_latest_run_for_default_uses_view_no_schema_filter() -> None:
|
||||
"""Дефолт (schema_version=None): читает v_analysis_runs_latest, без фильтра схемы."""
|
||||
db, captured = _mock_db_capturing()
|
||||
|
||||
row = latest_run_for(db, "66:41:0000000:1")
|
||||
|
||||
assert row is db._sentinel_row
|
||||
assert len(captured) == 1
|
||||
stmt, params = captured[0]
|
||||
upper = str(stmt).upper()
|
||||
assert "FROM V_ANALYSIS_RUNS_LATEST" in upper
|
||||
assert "SCHEMA_VERSION =" not in upper # дефолт НЕ фильтрует по схеме
|
||||
assert "CAST(:CAD_NUM AS TEXT)" in upper
|
||||
assert params == {"cad_num": "66:41:0000000:1"}
|
||||
|
||||
|
||||
def test_latest_run_for_schema_filtered_uses_base_table() -> None:
|
||||
"""schema_version задан: читает БАЗОВУЮ analysis_runs с фильтром + ORDER BY DESC LIMIT 1
|
||||
и прокидывает schema_version в params."""
|
||||
db, captured = _mock_db_capturing()
|
||||
|
||||
row = latest_run_for(db, "66:41:0000000:1", schema_version=ANALYZE_SCHEMA_VERSION)
|
||||
|
||||
assert row is db._sentinel_row
|
||||
assert len(captured) == 1
|
||||
stmt, params = captured[0]
|
||||
upper = str(stmt).upper()
|
||||
flat = " ".join(upper.split())
|
||||
# базовая таблица, НЕ view
|
||||
assert "FROM ANALYSIS_RUNS" in upper
|
||||
assert "V_ANALYSIS_RUNS_LATEST" not in upper
|
||||
# фильтр по схеме + latest-by-created_at
|
||||
assert "SCHEMA_VERSION = CAST(:SCHEMA_VERSION AS TEXT)" in upper
|
||||
assert "ORDER BY CREATED_AT DESC LIMIT 1" in flat
|
||||
assert "CAST(:CAD_NUM AS TEXT)" in upper
|
||||
# оба bind прокинуты
|
||||
assert params == {
|
||||
"cad_num": "66:41:0000000:1",
|
||||
"schema_version": ANALYZE_SCHEMA_VERSION,
|
||||
}
|
||||
|
||||
|
||||
def test_latest_run_for_no_double_colon_cast() -> None:
|
||||
"""psycopg v3 (оба режима): ни один bind не использует :name::type (recurring bug)."""
|
||||
db, captured = _mock_db_capturing()
|
||||
latest_run_for(db, "x")
|
||||
latest_run_for(db, "x", schema_version=ANALYZE_SCHEMA_VERSION)
|
||||
for stmt, _ in captured:
|
||||
sql = str(stmt)
|
||||
assert not re.search(r":[a-z_]+::[a-z]", sql), f"found :x::type cast in: {sql}"
|
||||
|
||||
|
||||
def test_latest_run_for_schema_filter_returns_filtered_not_newer_row() -> None:
|
||||
"""Поведенческий: при наличии более НОВОГО '1.0' рана фильтрованный режим должен
|
||||
вернуть последний 'analyze-1.0', а НЕ '1.0' (модель «view вернул бы §22, мы — base»).
|
||||
|
||||
Мокаем БД так, что фильтрованный запрос (по 'analyze-1.0') отдаёт analyze-строку,
|
||||
а запрос без фильтра (view, latest-overall) отдал бы более новую §22-строку. Так
|
||||
закрепляем смысл фичи: для базы форсайта берём именно analyze-ран, не свежий §22.
|
||||
"""
|
||||
analyze_row = MagicMock(name="analyze-1.0-row")
|
||||
analyze_row.schema_version = "analyze-1.0"
|
||||
forecast_row = MagicMock(name="1.0-row")
|
||||
forecast_row.schema_version = "1.0"
|
||||
|
||||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||||
upper = str(stmt).upper()
|
||||
result = MagicMock()
|
||||
if "FROM ANALYSIS_RUNS" in upper and "SCHEMA_VERSION =" in upper:
|
||||
# фильтрованный запрос: вернёт строку запрошенной схемы
|
||||
wanted = (params or {}).get("schema_version")
|
||||
result.first.return_value = analyze_row if wanted == "analyze-1.0" else forecast_row
|
||||
else:
|
||||
# view (без фильтра) — latest-overall = более новый §22-ран
|
||||
result.first.return_value = forecast_row
|
||||
return result
|
||||
|
||||
db = MagicMock()
|
||||
db.execute = execute_router
|
||||
|
||||
# фильтрованный режим → analyze-ран (не свежий §22)
|
||||
got = latest_run_for(db, "66:41:0000000:1", schema_version="analyze-1.0")
|
||||
assert got is analyze_row
|
||||
assert got.schema_version == "analyze-1.0"
|
||||
|
||||
# дефолт (view) → latest-overall = §22-ран (контраст, доказывает зачем фильтр)
|
||||
got_view = latest_run_for(db, "66:41:0000000:1")
|
||||
assert got_view is forecast_row
|
||||
255
backend/tests/workers/tasks/test_forecast.py
Normal file
255
backend/tests/workers/tasks/test_forecast.py
Normal file
|
|
@ -0,0 +1,255 @@
|
|||
"""Тесты Celery-таски forecast_site_finder_report (#3b-i, workers/tasks/forecast.py).
|
||||
|
||||
Чистые тесты (mock session + mock читателя/оркестратора/persist, без живой БД/§9.x).
|
||||
Покрывают:
|
||||
• happy path: read latest 'analyze-1.0' → build_site_finder_report → persist с
|
||||
schema_version='1.0', status='done', report-dict как result, верными
|
||||
confidence/segment/params; возврат нового id;
|
||||
• graceful: нет 'analyze-1.0' рана → return None, build НЕ зван, persist НЕ зван,
|
||||
без исключения;
|
||||
• read зовётся именно с schema_version=ANALYZE_SCHEMA_VERSION (база = site-analysis);
|
||||
• horizon домешивается в default-сетку горизонтов;
|
||||
• неожиданный сбой compute → return None (таска не валит воркер), db.close вызван.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from app.services.analysis_runs.repository import ANALYZE_SCHEMA_VERSION
|
||||
|
||||
|
||||
def _fake_analyze_row(
|
||||
*,
|
||||
result: dict[str, Any] | None = None,
|
||||
district: str | None = "Кировский",
|
||||
run_id: int = 101,
|
||||
) -> MagicMock:
|
||||
"""Fake Row последнего 'analyze-1.0' рана (атрибутный доступ row.result/.district/.id)."""
|
||||
row = MagicMock(name="analyze-row")
|
||||
row.id = run_id
|
||||
row.district = district
|
||||
row.result = (
|
||||
result
|
||||
if result is not None
|
||||
else {
|
||||
"district": {"district_name": "Кировский"},
|
||||
"competitors": [{"obj_class": "комфорт"}],
|
||||
}
|
||||
)
|
||||
return row
|
||||
|
||||
|
||||
def _fake_report_dict(
|
||||
*,
|
||||
schema_version: str = "1.0",
|
||||
confidence_level: str | None = "low",
|
||||
segment: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Стаб SiteFinderReport.as_dict() — только поля, которые читает таска для persist."""
|
||||
return {
|
||||
"schema_version": schema_version,
|
||||
"advisory": True,
|
||||
"confidence": {"level": confidence_level},
|
||||
"meta": {"segment": segment if segment is not None else {"obj_class": "комфорт"}},
|
||||
"exec_summary": {"headline": "stub"},
|
||||
}
|
||||
|
||||
|
||||
def _fake_report(d: dict[str, Any]) -> MagicMock:
|
||||
"""Стаб отчёта с as_dict() → d."""
|
||||
report = MagicMock(name="report")
|
||||
report.as_dict.return_value = d
|
||||
return report
|
||||
|
||||
|
||||
# ── happy path ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_task_happy_path_persists_forecast_run() -> None:
|
||||
"""Read 'analyze-1.0' → build → persist('1.0', status='done', report-dict, right
|
||||
confidence/segment/params); возвращает новый id."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
report_dict = _fake_report_dict(confidence_level="low", segment={"obj_class": "комфорт"})
|
||||
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=_fake_analyze_row()) as mock_read,
|
||||
patch.object(
|
||||
task_mod, "build_site_finder_report", return_value=_fake_report(report_dict)
|
||||
) as mock_build,
|
||||
patch.object(task_mod, "persist_analysis_run", return_value=777) as mock_persist,
|
||||
):
|
||||
result = task_mod.forecast_site_finder_report.run("66:41:0000000:1", 12, "tester")
|
||||
|
||||
assert result == 777
|
||||
|
||||
# read взят именно по site-analysis схеме
|
||||
mock_read.assert_called_once()
|
||||
assert mock_read.call_args.kwargs["schema_version"] == ANALYZE_SCHEMA_VERSION
|
||||
|
||||
# build получил analyze-dict + district + cad_num
|
||||
mock_build.assert_called_once()
|
||||
bkw = mock_build.call_args.kwargs
|
||||
assert bkw["cad_num"] == "66:41:0000000:1"
|
||||
assert bkw["district"] == "Кировский"
|
||||
assert bkw["analyze"]["competitors"] == [{"obj_class": "комфорт"}]
|
||||
assert 12 in bkw["horizons"]
|
||||
|
||||
# persist получил отчёт-dict как result + контракт §22-рана
|
||||
mock_persist.assert_called_once()
|
||||
pkw = mock_persist.call_args.kwargs
|
||||
assert pkw["cad_num"] == "66:41:0000000:1"
|
||||
assert pkw["result"] is report_dict
|
||||
assert pkw["schema_version"] == "1.0"
|
||||
assert pkw["status"] == "done"
|
||||
assert pkw["confidence"] == "low"
|
||||
assert pkw["segment"] == {"obj_class": "комфорт"}
|
||||
assert pkw["params"] == {"horizon": 12, "source": "forecast-task"}
|
||||
assert pkw["district"] == "Кировский"
|
||||
assert pkw["created_by"] == "tester"
|
||||
|
||||
db.close.assert_called_once()
|
||||
|
||||
|
||||
def test_task_district_fallback_to_analyze_dict() -> None:
|
||||
"""district из row.district = None → fallback в analyze['district']['district_name']."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
row = _fake_analyze_row(
|
||||
district=None,
|
||||
result={"district": {"district_name": "Чкаловский"}, "competitors": []},
|
||||
)
|
||||
report_dict = _fake_report_dict()
|
||||
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=row),
|
||||
patch.object(
|
||||
task_mod, "build_site_finder_report", return_value=_fake_report(report_dict)
|
||||
) as mock_build,
|
||||
patch.object(task_mod, "persist_analysis_run", return_value=5),
|
||||
):
|
||||
task_mod.forecast_site_finder_report.run("66:41:0000000:2", 12, None)
|
||||
|
||||
assert mock_build.call_args.kwargs["district"] == "Чкаловский"
|
||||
|
||||
|
||||
def test_task_horizon_merged_into_default_grid() -> None:
|
||||
"""Запрошенный horizon (вне default-сетки) домешивается в horizons, отсортировано."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=_fake_analyze_row()),
|
||||
patch.object(
|
||||
task_mod, "build_site_finder_report", return_value=_fake_report(_fake_report_dict())
|
||||
) as mock_build,
|
||||
patch.object(task_mod, "persist_analysis_run", return_value=1),
|
||||
):
|
||||
task_mod.forecast_site_finder_report.run("66:41:0000000:1", 9, None)
|
||||
|
||||
horizons = mock_build.call_args.kwargs["horizons"]
|
||||
assert 9 in horizons
|
||||
assert horizons == sorted(horizons)
|
||||
# default-сетка сохранена
|
||||
assert {6, 12, 18, 24}.issubset(set(horizons))
|
||||
|
||||
|
||||
# ── graceful: no base run ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_task_graceful_when_no_analyze_run() -> None:
|
||||
"""Нет 'analyze-1.0' рана → return None, build НЕ зван, persist НЕ зван, без исключения."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=None) as mock_read,
|
||||
patch.object(task_mod, "build_site_finder_report") as mock_build,
|
||||
patch.object(task_mod, "persist_analysis_run") as mock_persist,
|
||||
):
|
||||
result = task_mod.forecast_site_finder_report.run("66:41:0000000:9", 12, None)
|
||||
|
||||
assert result is None
|
||||
mock_read.assert_called_once()
|
||||
mock_build.assert_not_called()
|
||||
mock_persist.assert_not_called()
|
||||
db.close.assert_called_once()
|
||||
|
||||
|
||||
# ── resilience: compute blows up ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_task_returns_none_on_compute_error() -> None:
|
||||
"""Неожиданный сбой build_site_finder_report → return None (таска не валит воркер),
|
||||
persist НЕ зван, db закрыта."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=_fake_analyze_row()),
|
||||
patch.object(
|
||||
task_mod, "build_site_finder_report", side_effect=RuntimeError("§9.x blew up")
|
||||
),
|
||||
patch.object(task_mod, "persist_analysis_run") as mock_persist,
|
||||
):
|
||||
result = task_mod.forecast_site_finder_report.run("66:41:0000000:1", 12, None)
|
||||
|
||||
assert result is None
|
||||
mock_persist.assert_not_called()
|
||||
db.close.assert_called_once()
|
||||
|
||||
|
||||
def test_task_returns_none_when_persist_returns_none() -> None:
|
||||
"""persist best-effort вернул None (write провалился) → таска возвращает None, без краша."""
|
||||
from app.workers.tasks import forecast as task_mod
|
||||
|
||||
db = MagicMock()
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "latest_run_for", return_value=_fake_analyze_row()),
|
||||
patch.object(
|
||||
task_mod, "build_site_finder_report", return_value=_fake_report(_fake_report_dict())
|
||||
),
|
||||
patch.object(task_mod, "persist_analysis_run", return_value=None),
|
||||
):
|
||||
result = task_mod.forecast_site_finder_report.run("66:41:0000000:1", 12, None)
|
||||
|
||||
assert result is None
|
||||
db.close.assert_called_once()
|
||||
|
||||
|
||||
# ── registration / time limit ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_task_registered_in_celery_include() -> None:
|
||||
"""Таска зарегистрирована (module в include-списке celery_app)."""
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
assert "app.workers.tasks.forecast" in celery_app.conf.include
|
||||
|
||||
|
||||
def test_task_has_generous_time_limit() -> None:
|
||||
"""build_site_finder_report идёт минуты → щедрый per-task лимит (time_limit=900)."""
|
||||
from app.workers.tasks.forecast import forecast_site_finder_report
|
||||
|
||||
assert forecast_site_finder_report.time_limit == 900
|
||||
assert forecast_site_finder_report.soft_time_limit == 840
|
||||
|
||||
|
||||
def test_task_name_matches_convention() -> None:
|
||||
"""Имя таски следует house-style 'tasks.<module>.<func>'."""
|
||||
from app.workers.tasks.forecast import forecast_site_finder_report
|
||||
|
||||
assert forecast_site_finder_report.name == "tasks.forecast.forecast_site_finder_report"
|
||||
Loading…
Add table
Reference in a new issue