From 1db488ba998b2cdcfaf334b9aba6c50d335075e0 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Wed, 3 Jun 2026 01:45:30 +0500 Subject: [PATCH] feat(workers): supply-layers refresh task + weekly beat (#950 Step 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Celery task tasks.supply_layers_refresh computes L1/L2/L3 supply layers per Sverdlovsk district via supply_layers.compute_all_layers and UPSERTs into supply_layers (m.125) on uq_supply_layers_logical, updating all mutable cols (units/area/expected_online_date/confidence/method/computed_at), never the key. Coherent single date.today() snapshot_date across the batch; per-row SAVEPOINT (begin_nested) so a bad row/district can't abort the weekly batch; per-district guard. District set is data-driven (UNION of objective_lots.district and domrf_kn_objects.district_name region_cd=66), not hardcoded. Registered weekly Mon 06:00 МСК (after scrape_kn 04:15 that refreshes free_flats, the L2 source). Adds the task module to celery_app include (beat resolves by name). 11 unit tests asserting the ON CONFLICT contract + savepoint + coherent snapshot. --- backend/app/workers/beat_schedule.py | 17 + backend/app/workers/celery_app.py | 1 + .../workers/tasks/supply_layers_refresh.py | 219 +++++++++++ .../workers/test_supply_layers_refresh.py | 342 ++++++++++++++++++ 4 files changed, 579 insertions(+) create mode 100644 backend/app/workers/tasks/supply_layers_refresh.py create mode 100644 backend/tests/workers/test_supply_layers_refresh.py diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 2eff355a..58378bf7 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -305,4 +305,21 @@ def build_beat_schedule() -> dict: "options": {"queue": "celery"}, } + # Supply-layers пересчёт (#950 EPIC 6 Step5, GG-форсайт) → supply_layers (м.125). + # Детерминированно UPSERT-ит L1(open)/L2(hidden)/L3(future) по всем районам Свердл. + # обл. через supply_layers.compute_all_layers. ON CONFLICT идемпотентен в рамках + # одного snapshot_date (same-day re-run обновляет мутабельные колонки). + # + # Понедельник 06:00 МСК (Celery conf.timezone=Europe/Moscow → crontab в МСК). + # ORDERING ASSUMPTION: scrape_kn (наполняет domrf_kn_objects.free_flats, м.124 — + # источник L2 hidden) идёт Mon 04:15 МСК (settings.scrape_kn_cron='15 4 * * mon'). + # 06:00 даёт ~1.75 ч запаса, чтобы supply-layers считались по СВЕЖИМ free_flats. + # Если scrape_kn перенесут позже/он затянется — сдвинуть этот cron соответственно + # (или сделать chain). Задача техническая, не в job_settings (как cbr/poi/refresh). + schedule["supply-layers-refresh-weekly"] = { + "task": "tasks.supply_layers_refresh.supply_layers_refresh", + "schedule": _parse_cron("0 6 * * mon"), + "options": {"queue": "celery"}, + } + return schedule diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 9a747d2b..b9d909a0 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -61,6 +61,7 @@ celery_app = Celery( "app.workers.tasks.scrape_cadastre", "app.workers.tasks.ekburg_permits_sync", "app.workers.tasks.cbr_macro_sync", + "app.workers.tasks.supply_layers_refresh", ], ) celery_app.conf.timezone = "Europe/Moscow" diff --git a/backend/app/workers/tasks/supply_layers_refresh.py b/backend/app/workers/tasks/supply_layers_refresh.py new file mode 100644 index 00000000..b51bbf48 --- /dev/null +++ b/backend/app/workers/tasks/supply_layers_refresh.py @@ -0,0 +1,219 @@ +"""Celery task: пересчёт 3-слойного склада предложения → ``supply_layers`` (м.125). + +#950 (Site Finder v2 / GG-форсайт, EPIC 6 «3-layer склад предложения», ТЗ §9.3), +Step 5. Для каждого района Свердловской обл. вызывает детерминированный сервис +``supply_layers.compute_all_layers`` (L1 open / L2 hidden / L3 future) и UPSERT-ит +полученные ``SupplyLayerRow`` в таблицу ``supply_layers`` по логическому ключу +``uq_supply_layers_logical`` (layer, district_name, complex_id, obj_class, source, +snapshot_date). + +Детерминированно, без LLM — сервис read-only считает агрегаты, этот worker только +пишет. ON CONFLICT делает повторные прогоны идемпотентными в рамках одного дня +(re-run того же ``snapshot_date`` обновляет мутабельные колонки + computed_at). + +Контракт записи (КРИТИЧНО, флаг database-expert): на конфликте обновляем ВСЕ +мутабельные колонки (units/area/expected_online_date/confidence/method/computed_at), +иначе при ре-прогоне в таблице останутся stale-значения. Ключевые колонки (layer, +district_name, complex_id, obj_class, source, snapshot_date) НЕ трогаем. + +Mirror conventions (cbr_macro_sync.py / refresh_quarter_price_index.py): + • ``SessionLocal()`` + try/finally close, ``logger`` (не print). + • SAVEPOINT per-row (``with db.begin_nested():``, backend.md) — одна битая строка + (напр. CHECK-violation) не откатывает весь батч; commit один раз в конце. + • ``CAST(:x AS type)`` — НИКОГДА ``:x::type`` (psycopg v3). + • Сбой одного района (сервис уже degrade'ит до [], но страхуемся) не валит батч. + +Coherent snapshot: ``run_date = date.today()`` вычисляется ОДИН раз в начале и +передаётся как ``snapshot_date`` КАЖДОЙ строке всех районов — весь батч делит один +снапшот (нет split на midnight-boundary; ON CONFLICT идемпотентен для same-day +re-run). Расписание — еженедельно, регистрируется в beat_schedule.py +(supply-layers-refresh-weekly). +""" + +from __future__ import annotations + +import logging +from datetime import date +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal +from app.services.site_finder.supply_layers import ( + SupplyLayerRow, + compute_all_layers, +) +from app.workers.celery_app import celery_app + +logger = logging.getLogger(__name__) + +# Регион данных (ЕКБ / Свердловская обл.). domrf_kn_objects.district_name заполнен +# spatial-join'ом только для region_cd=66 (м.57) — зеркало supply_layers._EKB_REGION_CODE. +_EKB_REGION_CODE: int = 66 + +# Множество районов для прогона = UNION distinct non-NULL районов из ОБОИХ источников +# (objective_lots.district + domrf_kn_objects.district_name WHERE region_cd=66). Не +# хардкодим 8 имён — робастно к новым районам. CAST(:x AS type) (psycopg v3). +_DISTRICTS_SQL = text( + """ + SELECT district_name FROM ( + SELECT DISTINCT ol.district AS district_name + FROM objective_lots ol + WHERE ol.district IS NOT NULL + UNION + SELECT DISTINCT o.district_name AS district_name + FROM domrf_kn_objects o + WHERE o.district_name IS NOT NULL + AND o.region_cd = CAST(:region_cd AS integer) + ) d + ORDER BY district_name + """ +) + +# UPSERT в supply_layers (м.125). Конфликт по uq_supply_layers_logical (NULLS NOT +# DISTINCT) → ON CONFLICT по точному column-list ключа. DO UPDATE обновляет ВСЕ +# мутабельные колонки + computed_at=now() (database-expert: иначе stale-данные). +# Ключевые колонки (layer/district_name/complex_id/obj_class/source/snapshot_date) +# НЕ обновляются — это и есть ключ. CAST(:x AS type) везде (psycopg v3, никогда ::). +_UPSERT_SQL = text( + """ + INSERT INTO supply_layers ( + layer, district_name, complex_id, obj_class, + units_estimate, area_estimate, expected_online_date, + source, confidence, method, snapshot_date + ) VALUES ( + CAST(:layer AS smallint), + CAST(:district_name AS text), + CAST(:complex_id AS bigint), + CAST(:obj_class AS text), + CAST(:units_estimate AS integer), + CAST(:area_estimate AS numeric), + CAST(:expected_online_date AS date), + CAST(:source AS text), + CAST(:confidence AS text), + CAST(:method AS text), + CAST(:snapshot_date AS date) + ) + ON CONFLICT (layer, district_name, complex_id, obj_class, source, snapshot_date) + DO UPDATE SET + units_estimate = EXCLUDED.units_estimate, + area_estimate = EXCLUDED.area_estimate, + expected_online_date = EXCLUDED.expected_online_date, + confidence = EXCLUDED.confidence, + method = EXCLUDED.method, + computed_at = now() + """ +) + + +def _list_districts(db: Session) -> list[str]: + """Distinct non-NULL районы из objective_lots + domrf_kn_objects (region_cd=66). + + UNION обоих источников — робастно к новым районам (не хардкодим 8 имён). На сбое + запроса логируем и возвращаем [] (батч завершится no-op, не упадёт). + """ + try: + rows = db.execute(_DISTRICTS_SQL, {"region_cd": _EKB_REGION_CODE}).mappings().all() + except Exception: + logger.exception("supply_layers_refresh: districts enumeration query failed") + return [] + return [r["district_name"] for r in rows if r["district_name"]] + + +def _upsert_rows(db: Session, rows: list[SupplyLayerRow]) -> tuple[int, int]: + """UPSERT строк слоёв в supply_layers. SAVEPOINT per-row (backend.md): одна битая + строка (напр. CHECK-violation) не откатывает весь батч. + + Returns: (upserted, skipped) — upserted = успешные INSERT/UPDATE, skipped = строки, + чей UPSERT упал (залогированы warning'ом). Commit делает caller (один на батч). + """ + upserted = 0 + skipped = 0 + for row in rows: + params = row.as_dict() + try: + with db.begin_nested(): + db.execute(_UPSERT_SQL, params) + upserted += 1 + except Exception as e: + skipped += 1 + logger.warning( + "supply_layers upsert failed (layer=%s district=%s source=%s): %s", + row.layer, + row.district_name, + row.source, + e, + ) + return upserted, skipped + + +@celery_app.task( + bind=True, + name="tasks.supply_layers_refresh.supply_layers_refresh", + max_retries=2, +) +def supply_layers_refresh(self: Any) -> dict[str, Any]: + """Пересчитать 3-слойный склад предложения по всем районам и UPSERT в supply_layers. + + Coherent snapshot: ``run_date = date.today()`` фиксируется ОДИН раз и передаётся + как ``snapshot_date`` каждой строке всех районов (избегает midnight-split, делает + same-day re-run идемпотентным). Цикл по районам: сбой/пустота сервиса одного района + не валит остальные (сервис уже degrade'ит до [], плюс per-district guard). Commit + один раз в конце (outer tx); per-row SAVEPOINT изолирует битые строки. + + Returns: + Счётчики ``{"snapshot_date", "districts", "rows", "upserted", "skipped"}``. + + Raises: + Пробрасывает только инфраструктурные сбои (commit / нефатальные внутри уже + обработаны) — чтобы фейл был виден в Celery / GlitchTip, а не молча проглочен. + """ + run_date = date.today() + db = SessionLocal() + try: + districts = _list_districts(db) + logger.info( + "supply_layers_refresh: %d districts to process (snapshot_date=%s)", + len(districts), + run_date.isoformat(), + ) + + all_rows: list[SupplyLayerRow] = [] + for district in districts: + try: + rows = compute_all_layers(db, district=district, snapshot_date=run_date) + except Exception: + # Сервис уже graceful (→ []), но страхуемся: сбой одного района не + # должен ронять весь батч. Логируем и продолжаем со следующим. + logger.exception( + "supply_layers_refresh: compute_all_layers failed for district=%s", + district, + ) + continue + all_rows.extend(rows) + + upserted, skipped = _upsert_rows(db, all_rows) + db.commit() + + logger.info( + "supply_layers_refresh: done snapshot_date=%s districts=%d rows=%d " + "upserted=%d skipped=%d", + run_date.isoformat(), + len(districts), + len(all_rows), + upserted, + skipped, + ) + return { + "snapshot_date": run_date.isoformat(), + "districts": len(districts), + "rows": len(all_rows), + "upserted": upserted, + "skipped": skipped, + } + except Exception as e: + logger.exception("supply_layers_refresh failed: %s", e) + raise + finally: + db.close() diff --git a/backend/tests/workers/test_supply_layers_refresh.py b/backend/tests/workers/test_supply_layers_refresh.py new file mode 100644 index 00000000..b82278b6 --- /dev/null +++ b/backend/tests/workers/test_supply_layers_refresh.py @@ -0,0 +1,342 @@ +"""Тесты Celery-таски supply_layers_refresh (#950 EPIC 6 Step5). + +Чистые тесты (mock session + mock сервиса, без живой сети/БД). Покрывают: + • контракт UPSERT-SQL: точный ON CONFLICT target (uq_supply_layers_logical column + list), DO UPDATE по ВСЕМ мутабельным колонкам (database-expert: иначе stale), + ключевые колонки НЕ обновляются, CAST(:x AS type) не :x::type (psycopg v3); + • district-enumeration: UNION objective_lots + domrf_kn_objects (region_cd=66), + DISTINCT non-NULL, форма запроса; + • coherent single snapshot_date: один date.today() прокинут КАЖДОЙ строке; + • per-row SAVEPOINT (db.begin_nested) на каждой строке; + • счётчики upserted/skipped; resilience: сбой compute одного района не валит батч. +""" + +from __future__ import annotations + +import os +from datetime import date +from typing import Any +from unittest.mock import MagicMock, call, patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services.site_finder.supply_layers import SupplyLayerRow + + +def _make_mock_db( + district_rows: list[dict[str, Any]], +) -> tuple[MagicMock, list[dict[str, Any]], list[Any]]: + """Mock DB session с поддержкой db.begin_nested() (SAVEPOINT). + + Первый db.execute (без params, либо с region_cd) трактуем как districts-запрос → + возвращаем district_rows через .mappings().all(). Остальные execute (UPSERT) — + захватываем params в captured_upserts. + + Returns: (db, captured_upserts, executed_sql) — captured_upserts содержит params + каждого UPSERT-execute; executed_sql — все stmt-объекты (для проверки SQL). + """ + captured_upserts: list[dict[str, Any]] = [] + executed_sql: list[Any] = [] + state = {"districts_served": False} + + def execute_router(stmt: Any, params: Any = None) -> MagicMock: + executed_sql.append(stmt) + result = MagicMock() + # districts-запрос: первый execute (region_cd-only params или до UPSERT'ов) + if not state["districts_served"] and "supply_layers" not in str(stmt).lower(): + state["districts_served"] = True + result.mappings.return_value.all.return_value = district_rows + return result + if params is not None: + captured_upserts.append(params) + return result + + db = MagicMock() + db.execute = execute_router + db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + return db, captured_upserts, executed_sql + + +def _row( + *, + layer: int, + district: str, + source: str, + confidence: str = "high", + units: int | None = 10, + area: float | None = 500.0, + obj_class: str | None = None, + expected: date | None = None, + snapshot_date: date | None = None, +) -> SupplyLayerRow: + return SupplyLayerRow( + layer=layer, # type: ignore[arg-type] + district_name=district, + complex_id=None, + obj_class=obj_class, + units_estimate=units, + area_estimate=area, + expected_online_date=expected, + source=source, # type: ignore[arg-type] + confidence=confidence, # type: ignore[arg-type] + method="test", + snapshot_date=snapshot_date, + ) + + +# ── UPSERT-SQL контракт ────────────────────────────────────────────────────── + + +def test_upsert_sql_conflict_target_and_updated_columns() -> None: + """ON CONFLICT таргетит точный логический ключ; DO UPDATE обновляет ВСЕ + мутабельные колонки + computed_at; ключевые колонки НЕ в SET; нет :x::type.""" + from app.workers.tasks.supply_layers_refresh import _UPSERT_SQL + + sql = str(_UPSERT_SQL) + upper = sql.upper() + + assert "INSERT INTO SUPPLY_LAYERS" in upper + # точный ON CONFLICT target = uq_supply_layers_logical column-list + assert ( + "ON CONFLICT (LAYER, DISTRICT_NAME, COMPLEX_ID, OBJ_CLASS, SOURCE, SNAPSHOT_DATE)" in upper + ) + assert "DO UPDATE SET" in upper + # ВСЕ мутабельные колонки обновляются (database-expert: иначе stale-данные) + for col in ( + "UNITS_ESTIMATE = EXCLUDED.UNITS_ESTIMATE", + "AREA_ESTIMATE = EXCLUDED.AREA_ESTIMATE", + "EXPECTED_ONLINE_DATE = EXCLUDED.EXPECTED_ONLINE_DATE", + "CONFIDENCE = EXCLUDED.CONFIDENCE", + "METHOD = EXCLUDED.METHOD", + "COMPUTED_AT = NOW()", + ): + assert col in upper, f"missing updated column: {col}" + # ключевые колонки НЕ обновляются в SET-части (после DO UPDATE SET) + set_clause = upper.split("DO UPDATE SET", 1)[1] + for key_col in ( + "LAYER = EXCLUDED", + "DISTRICT_NAME = EXCLUDED", + "COMPLEX_ID = EXCLUDED", + "OBJ_CLASS = EXCLUDED", + "SOURCE = EXCLUDED", + "SNAPSHOT_DATE = EXCLUDED", + ): + assert key_col not in set_clause, f"key column must not be updated: {key_col}" + # psycopg v3: CAST(:x AS type), НИКОГДА :x::type + assert "::" not in sql + assert "CAST(:LAYER AS SMALLINT)" in upper + assert "CAST(:SNAPSHOT_DATE AS DATE)" in upper + + +def test_no_double_colon_cast_in_sql() -> None: + """psycopg v3: ни один bind не использует :name::type (recurring bug class).""" + import re + + from app.workers.tasks.supply_layers_refresh import _DISTRICTS_SQL, _UPSERT_SQL + + for stmt in (_UPSERT_SQL, _DISTRICTS_SQL): + sql = str(stmt) + assert not re.search(r":[a-z_]+::[a-z]", sql), f"found :x::type cast in: {sql}" + + +# ── district enumeration ───────────────────────────────────────────────────── + + +def test_districts_query_unions_both_sources() -> None: + """Запрос районов = UNION objective_lots.district + domrf_kn_objects.district_name + (region_cd=66), DISTINCT non-NULL — не хардкодит имена.""" + from app.workers.tasks.supply_layers_refresh import _DISTRICTS_SQL + + sql = str(_DISTRICTS_SQL) + upper = sql.upper() + + assert "FROM OBJECTIVE_LOTS" in upper + assert "FROM DOMRF_KN_OBJECTS" in upper + assert "UNION" in upper + assert "DISTINCT" in upper + assert "IS NOT NULL" in upper + # region-фильтр domrf через CAST(:x AS integer) + assert "REGION_CD = CAST(:REGION_CD AS INTEGER)" in upper + + +def test_list_districts_filters_null_and_returns_names() -> None: + """_list_districts отдаёт только non-NULL имена; на пустом источнике → [].""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db, _, _ = _make_mock_db( + [{"district_name": "Кировский"}, {"district_name": None}, {"district_name": "Чкаловский"}] + ) + assert task_mod._list_districts(db) == ["Кировский", "Чкаловский"] + + db_empty, _, _ = _make_mock_db([]) + assert task_mod._list_districts(db_empty) == [] + + +def test_list_districts_graceful_on_query_error() -> None: + """Сбой districts-запроса → [] (батч no-op, не падает).""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db = MagicMock() + db.execute.side_effect = RuntimeError("db down") + assert task_mod._list_districts(db) == [] + + +# ── per-row UPSERT / SAVEPOINT ─────────────────────────────────────────────── + + +def test_upsert_rows_uses_savepoint_per_row() -> None: + """Каждая строка UPSERT-ится внутри db.begin_nested() (SAVEPOINT, backend.md).""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db, captured, _ = _make_mock_db([]) + # пометим districts как обслуженные, чтобы execute шёл в UPSERT-ветку + db.execute("SELECT FROM supply_layers warmup", None) + snap = date(2026, 6, 3) + rows = [ + _row(layer=1, district="Кировский", source="objective", snapshot_date=snap), + _row(layer=2, district="Кировский", source="domrf_declared", snapshot_date=snap), + ] + upserted, skipped = task_mod._upsert_rows(db, rows) + + assert upserted == 2 + assert skipped == 0 + # begin_nested вызван по разу на строку + assert db.begin_nested.call_count == 2 + # params каждого UPSERT — это row.as_dict() + assert captured == [rows[0].as_dict(), rows[1].as_dict()] + + +def test_upsert_rows_savepoint_isolates_bad_row() -> None: + """Битая строка (execute бросает) → skipped++, остальные продолжают (SAVEPOINT).""" + from app.workers.tasks import supply_layers_refresh as task_mod + + captured: list[dict[str, Any]] = [] + calls = {"n": 0} + + def execute_router(stmt: Any, params: Any = None) -> MagicMock: + calls["n"] += 1 + if calls["n"] == 2: # вторая строка — битая + raise RuntimeError("CHECK violation") + captured.append(params) + return MagicMock() + + db = MagicMock() + db.execute = execute_router + db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + + rows = [ + _row(layer=1, district="A", source="objective", snapshot_date=date(2026, 6, 3)), + _row(layer=2, district="B", source="domrf_declared", snapshot_date=date(2026, 6, 3)), + _row(layer=3, district="C", source="domrf_multiphase", snapshot_date=date(2026, 6, 3)), + ] + upserted, skipped = task_mod._upsert_rows(db, rows) + + assert upserted == 2 + assert skipped == 1 + # первая и третья строки прошли, вторая — нет + assert [p["district_name"] for p in captured] == ["A", "C"] + + +# ── end-to-end task: coherent snapshot, counters, resilience ───────────────── + + +def test_task_coherent_single_snapshot_date() -> None: + """run_date = date.today() ОДИН раз → передан snapshot_date КАЖДОЙ строке всех + районов; ВСЕ UPSERT-params несут один и тот же snapshot_date.""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db, captured, _ = _make_mock_db( + [{"district_name": "Кировский"}, {"district_name": "Чкаловский"}] + ) + today = date(2026, 6, 3) + + def fake_compute(_db: Any, *, district: str, snapshot_date: date) -> list[SupplyLayerRow]: + # сервис обязан получить наш единый snapshot_date + assert snapshot_date == today + return [_row(layer=1, district=district, source="objective", snapshot_date=snapshot_date)] + + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch.object(task_mod, "date") as mock_date, + patch.object(task_mod, "compute_all_layers", side_effect=fake_compute) as mock_compute, + ): + mock_date.today.return_value = today + result = task_mod.supply_layers_refresh.run() + + # compute вызван на каждый район с единым snapshot_date + assert mock_compute.call_count == 2 + for c in mock_compute.call_args_list: + assert c == call(db, district=c.kwargs["district"], snapshot_date=today) + # все UPSERT-params несут один snapshot_date + assert captured, "expected at least one upsert" + assert {p["snapshot_date"] for p in captured} == {today} + assert result["snapshot_date"] == today.isoformat() + assert result["districts"] == 2 + assert result["rows"] == 2 + assert result["upserted"] == 2 + assert result["skipped"] == 0 + db.commit.assert_called_once() + db.close.assert_called_once() + + +def test_task_resilient_to_one_district_failure() -> None: + """Сбой compute_all_layers одного района не валит батч: остальные районы + обрабатываются, commit вызывается.""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db, captured, _ = _make_mock_db([{"district_name": "Good"}, {"district_name": "Bad"}]) + + def fake_compute(_db: Any, *, district: str, snapshot_date: date) -> list[SupplyLayerRow]: + if district == "Bad": + raise RuntimeError("service blew up") + return [_row(layer=1, district=district, source="objective", snapshot_date=snapshot_date)] + + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch.object(task_mod, "compute_all_layers", side_effect=fake_compute), + ): + result = task_mod.supply_layers_refresh.run() + + # только "Good" дал строку; "Bad" проглочен (logged) и пропущен + assert result["districts"] == 2 + assert result["rows"] == 1 + assert result["upserted"] == 1 + assert [p["district_name"] for p in captured] == ["Good"] + db.commit.assert_called_once() + db.close.assert_called_once() + + +def test_task_closes_db_on_infra_error() -> None: + """Инфраструктурный сбой (напр. districts-enum) пробрасывается, db закрывается.""" + from app.workers.tasks import supply_layers_refresh as task_mod + + db = MagicMock() + db.execute.side_effect = None + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch.object(task_mod, "_list_districts", side_effect=RuntimeError("boom")), + ): + try: + task_mod.supply_layers_refresh.run() + raise AssertionError("expected RuntimeError to propagate") + except RuntimeError as e: + assert "boom" in str(e) + + db.close.assert_called_once() + + +# ── beat registration ──────────────────────────────────────────────────────── + + +def test_task_registered_in_beat_weekly_monday() -> None: + """Задача зарегистрирована в beat как еженедельная (понедельник).""" + from app.workers.beat_schedule import build_beat_schedule + + schedule = build_beat_schedule() + assert "supply-layers-refresh-weekly" in schedule + entry = schedule["supply-layers-refresh-weekly"] + assert entry["task"] == "tasks.supply_layers_refresh.supply_layers_refresh" + # crontab day_of_week = понедельник (1) + assert 1 in entry["schedule"].day_of_week