fix(workers): count lots_pf failure correctly, no silent done (#1220) #1267
2 changed files with 214 additions and 2 deletions
|
|
@ -263,8 +263,14 @@ def sync_objective_group(
|
|||
use_dkp=params.get("use_dkp", False),
|
||||
payload=None,
|
||||
)
|
||||
n_requests += 1
|
||||
reports_ok += 1
|
||||
# NB: n_requests / reports_ok инкрементируем ТОЛЬКО ПОСЛЕ
|
||||
# успешного завершения stream+parse+commit. Иначе при провале
|
||||
# внутреннего generic-except (issue #1220) счётчики бы росли
|
||||
# даже на упавшем главном отчёте, и _finish_run помечал бы
|
||||
# run status='done' → silent failure для downstream.
|
||||
# ObjectiveAuthError/ObjectiveAPIError ⊂ RuntimeError ⊂ Exception
|
||||
# → ловятся именно этим inner-except (внешний (Auth|API)Error-блок
|
||||
# на строке ~355 для lots_pf недостижим).
|
||||
try:
|
||||
with client.stream_report(
|
||||
report_type="Поквартирные",
|
||||
|
|
@ -290,8 +296,12 @@ def sync_objective_group(
|
|||
{"n": n_lots, "rid": raw_id},
|
||||
)
|
||||
db.commit()
|
||||
n_requests += 1
|
||||
reports_ok += 1
|
||||
except Exception as parse_err:
|
||||
db.rollback()
|
||||
n_requests += 1
|
||||
reports_failed += 1
|
||||
logger.exception(
|
||||
"sync_objective_group: lots_pf stream failed raw_id=%s: %s",
|
||||
raw_id,
|
||||
|
|
|
|||
202
backend/tests/workers/tasks/test_scrape_objective.py
Normal file
202
backend/tests/workers/tasks/test_scrape_objective.py
Normal file
|
|
@ -0,0 +1,202 @@
|
|||
"""Тесты для backend/app/workers/tasks/scrape_objective.py — fix P2 silent-failure
|
||||
из issue #1220.
|
||||
|
||||
Покрытие:
|
||||
- lots_pf success-path: `reports_ok=1, reports_failed=0` (старое поведение сохранено);
|
||||
- lots_pf бросает `ObjectiveAuthError` (⊂ RuntimeError ⊂ Exception): теперь
|
||||
`reports_failed=1, reports_ok=0` — БЫЛО `reports_ok=1, reports_failed=0` (silent
|
||||
done на провале главного отчёта);
|
||||
- lots_pf бросает `ObjectiveAPIError`: симметрично — `reports_failed=1, reports_ok=0`;
|
||||
- lots_pf бросает generic `RuntimeError`: тот же graceful-fail + counter;
|
||||
- `_finish_run` при `reports_failed>0` помечает run `status='failed'`, не `'done'`.
|
||||
|
||||
Перед патчем счётчик `reports_ok += 1` стоял ДО `with client.stream_report(...)` блока
|
||||
и не откатывался во внутреннем `except Exception` — отчёт «Лоты» (~600МБ) падал,
|
||||
а run помечался `status='done'`. Сосед `corp_sum` уже считал корректно (после save_raw).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import ModuleType
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_fake_parser(parse_lots_pf_stream: Any, parse_corp_sum: Any) -> Any:
|
||||
"""Возвращает callable spec_from_file_location-подобный объект, который
|
||||
подменяет загрузку 70_parse_objective_raw.py и впрыскивает fake-parser.
|
||||
"""
|
||||
|
||||
def fake_spec_from_file_location(_name: str, _path: Any) -> Any:
|
||||
spec = MagicMock()
|
||||
|
||||
class _Loader:
|
||||
def exec_module(self, module: ModuleType) -> None:
|
||||
module.parse_lots_pf_stream = parse_lots_pf_stream
|
||||
module.parse_corp_sum = parse_corp_sum
|
||||
|
||||
spec.loader = _Loader()
|
||||
return spec
|
||||
|
||||
def fake_module_from_spec(_spec: Any) -> ModuleType:
|
||||
return ModuleType("objective_parser_70_fake")
|
||||
|
||||
return fake_spec_from_file_location, fake_module_from_spec
|
||||
|
||||
|
||||
def _make_db_mock() -> MagicMock:
|
||||
"""Mock SessionLocal() — все execute().scalar_one() возвращают целые ID."""
|
||||
db = MagicMock()
|
||||
# _start_run → INSERT ... RETURNING run_id
|
||||
# _save_raw → INSERT ... RETURNING raw_id
|
||||
db.execute.return_value.scalar_one.return_value = 42
|
||||
return db
|
||||
|
||||
|
||||
class _FakeStreamCtx:
|
||||
"""Контекст-менеджер, имитирующий client.stream_report(...) → resp.iter_bytes."""
|
||||
|
||||
def __init__(self, exc: Exception | None = None) -> None:
|
||||
self._exc = exc
|
||||
self.iter_bytes = MagicMock(return_value=iter([b"{}"]))
|
||||
|
||||
def __enter__(self) -> Any:
|
||||
if self._exc is not None:
|
||||
raise self._exc
|
||||
return self
|
||||
|
||||
def __exit__(self, *a: Any) -> None:
|
||||
return None
|
||||
|
||||
|
||||
def _patched_run(
|
||||
*,
|
||||
stream_exc: Exception | None,
|
||||
parse_lots_fn: Any = None,
|
||||
) -> tuple[dict[str, Any], list[Any]]:
|
||||
"""Запускает sync_objective_group.run() с замоканными:
|
||||
SessionLocal, ObjectiveClient, parser_mod, settings.
|
||||
|
||||
Возвращает (return-value task, [список _finish_run kwargs]) для assert'ов.
|
||||
"""
|
||||
from app.workers.tasks import scrape_objective as mod
|
||||
|
||||
db = _make_db_mock()
|
||||
client = MagicMock()
|
||||
client.stream_report = MagicMock(return_value=_FakeStreamCtx(exc=stream_exc))
|
||||
# corp_sum ветка — успешно
|
||||
client.report_corpuses_summary = MagicMock(return_value={"some": "payload"})
|
||||
|
||||
parse_lots = parse_lots_fn or (lambda *a, **kw: (10, 20)) # n_lots, n_hist
|
||||
parse_corp = MagicMock(return_value=5)
|
||||
|
||||
fake_spec, fake_mod_from_spec = _make_fake_parser(parse_lots, parse_corp)
|
||||
|
||||
finish_calls: list[dict[str, Any]] = []
|
||||
orig_finish = mod._finish_run
|
||||
|
||||
def capture_finish(_db: Any, _run_id: int, **kwargs: Any) -> None:
|
||||
finish_calls.append(kwargs)
|
||||
orig_finish(_db, _run_id, **kwargs)
|
||||
|
||||
with (
|
||||
patch.object(mod, "SessionLocal", return_value=db),
|
||||
patch.object(mod, "ObjectiveClient", return_value=client),
|
||||
patch("importlib.util.spec_from_file_location", side_effect=fake_spec),
|
||||
patch("importlib.util.module_from_spec", side_effect=fake_mod_from_spec),
|
||||
patch.object(mod.settings, "objective_api_key", "test-key"),
|
||||
patch.object(mod.settings, "objective_default_group", "Свердловская область"),
|
||||
patch.object(mod, "_finish_run", side_effect=capture_finish),
|
||||
):
|
||||
result = mod.sync_objective_group.run(
|
||||
group_name="Свердловская область",
|
||||
triggered_by="manual",
|
||||
)
|
||||
return result, finish_calls
|
||||
|
||||
|
||||
# ── tests ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_lots_pf_success_keeps_reports_ok_one_failed_zero() -> None:
|
||||
"""Success-path: stream+parse прошли → reports_ok=1, reports_failed=0
|
||||
(старое поведение, регресс-страховка).
|
||||
"""
|
||||
result, finish_calls = _patched_run(stream_exc=None)
|
||||
|
||||
# corp_sum + lots_pf оба успешны → reports_ok=2, reports_failed=0
|
||||
assert result["reports_ok"] == 2
|
||||
assert result["reports_failed"] == 0
|
||||
assert result["rows_lots"] == 10
|
||||
assert result["rows_history"] == 20
|
||||
assert result["rows_corpus_room"] == 5
|
||||
|
||||
assert len(finish_calls) == 1
|
||||
assert finish_calls[0]["status"] == "done"
|
||||
assert finish_calls[0]["error"] is None
|
||||
assert finish_calls[0]["reports_ok"] == 2
|
||||
assert finish_calls[0]["reports_failed"] == 0
|
||||
|
||||
|
||||
def test_lots_pf_auth_error_increments_failed_not_ok() -> None:
|
||||
"""ObjectiveAuthError ⊂ RuntimeError ⊂ Exception — ловится во внутреннем
|
||||
generic-except (внешний (Auth|API)Error-блок недостижим). Должен НЕ
|
||||
инкрементить reports_ok и инкрементить reports_failed.
|
||||
|
||||
Регресс fix #1220: до патча reports_ok=1, reports_failed=0 → status='done'.
|
||||
"""
|
||||
from app.services.scrapers.objective import ObjectiveAuthError
|
||||
|
||||
result, finish_calls = _patched_run(stream_exc=ObjectiveAuthError("HTTP 401"))
|
||||
|
||||
# corp_sum успешен (reports_ok=1), lots_pf упал (reports_failed=1)
|
||||
assert result["reports_ok"] == 1
|
||||
assert result["reports_failed"] == 1
|
||||
assert result["rows_lots"] == 0
|
||||
assert result["rows_history"] == 0
|
||||
|
||||
assert finish_calls[0]["status"] == "failed"
|
||||
assert finish_calls[0]["error"] is not None
|
||||
assert finish_calls[0]["reports_ok"] == 1
|
||||
assert finish_calls[0]["reports_failed"] == 1
|
||||
|
||||
|
||||
def test_lots_pf_api_error_increments_failed_not_ok() -> None:
|
||||
"""ObjectiveAPIError (HTTP 500/429/etc) — то же поведение, что Auth."""
|
||||
from app.services.scrapers.objective import ObjectiveAPIError
|
||||
|
||||
result, finish_calls = _patched_run(stream_exc=ObjectiveAPIError("HTTP 500"))
|
||||
|
||||
assert result["reports_ok"] == 1 # только corp_sum
|
||||
assert result["reports_failed"] == 1
|
||||
assert finish_calls[0]["status"] == "failed"
|
||||
|
||||
|
||||
def test_lots_pf_generic_exception_increments_failed_not_ok() -> None:
|
||||
"""Generic RuntimeError из parser_mod.parse_lots_pf_stream — также
|
||||
graceful-fail, считается как reports_failed."""
|
||||
|
||||
def _boom(*_a: Any, **_kw: Any) -> tuple[int, int]:
|
||||
raise RuntimeError("ijson parse exploded mid-stream")
|
||||
|
||||
result, finish_calls = _patched_run(stream_exc=None, parse_lots_fn=_boom)
|
||||
|
||||
assert result["reports_ok"] == 1 # corp_sum ОК
|
||||
assert result["reports_failed"] == 1
|
||||
assert result["rows_lots"] == 0
|
||||
assert finish_calls[0]["status"] == "failed"
|
||||
assert "1 reports failed" in (finish_calls[0]["error"] or "")
|
||||
|
||||
|
||||
def test_finish_run_status_failed_when_any_report_failed() -> None:
|
||||
"""Контракт _finish_run: при reports_failed>0 status='failed', не 'done'.
|
||||
error содержит '<n> reports failed'.
|
||||
"""
|
||||
from app.services.scrapers.objective import ObjectiveAuthError
|
||||
|
||||
_, finish_calls = _patched_run(stream_exc=ObjectiveAuthError("nope"))
|
||||
|
||||
assert finish_calls[0]["status"] == "failed"
|
||||
assert finish_calls[0]["error"] == "1 reports failed"
|
||||
Loading…
Add table
Reference in a new issue