diff --git a/backend/app/services/analysis_runs/repository.py b/backend/app/services/analysis_runs/repository.py index 18d6bc89..5e4b5cdb 100644 --- a/backend/app/services/analysis_runs/repository.py +++ b/backend/app/services/analysis_runs/repository.py @@ -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() diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index b9d909a0..59189d4d 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -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" diff --git a/backend/app/workers/tasks/forecast.py b/backend/app/workers/tasks/forecast.py new file mode 100644 index 00000000..f7a1243b --- /dev/null +++ b/backend/app/workers/tasks/forecast.py @@ -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() diff --git a/backend/tests/services/analysis_runs/__init__.py b/backend/tests/services/analysis_runs/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/services/analysis_runs/test_repository_latest_run.py b/backend/tests/services/analysis_runs/test_repository_latest_run.py new file mode 100644 index 00000000..6612a56d --- /dev/null +++ b/backend/tests/services/analysis_runs/test_repository_latest_run.py @@ -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 diff --git a/backend/tests/workers/tasks/test_forecast.py b/backend/tests/workers/tasks/test_forecast.py new file mode 100644 index 00000000..167debd9 --- /dev/null +++ b/backend/tests/workers/tasks/test_forecast.py @@ -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..'.""" + from app.workers.tasks.forecast import forecast_site_finder_report + + assert forecast_site_finder_report.name == "tasks.forecast.forecast_site_finder_report"