Light1YT
492574fe17
All checks were successful
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 6m26s
CI / backend-tests (pull_request) Successful in 6m27s
Два concurrency-бага в scrape_kn: 1. Lock value="1" без owner-токена + TTL=30мин = заявленной длительности sweep'а (нулевой запас). Sweep > TTL → lock истекает, второй sweep стартует, первый при завершении безусловным r.delete сносит ЧУЖОЙ lock → возможен третий параллельный. 2. SIGKILL/redeploy mid-sweep: finally не выполняется, lock живёт до 30 мин. worker_ready метит run 'zombie' и enqueue'ит resume_kn_run БЕЗ countdown, который ловит lock_held и возвращает skipped без retry → run теряется до недельного beat. Patch: - _region_lock: value = uuid4().hex; release через Lua check-and-delete (`if GET == ARGV[1] then DEL else 0 end`) — не сносим чужой lock. - _LOCK_TTL_SECONDS 30 → 45 мин (полтора max sweep duration, запас). - resume_kn_run: при lock_held → raise self.retry(countdown=300, max_retries=12) → ~час окно для подхвата вместо silent skip. - scrape_kn_region (scheduled) намеренно остаётся skipped — beat поднимет в следующий weekly tick (другая семантика). 11 новых юнит-тестов (token uniqueness, Lua guard, retry-not-skip, existing behaviors). 16/16 scrape_kn тестов зелёные. ruff clean. Closes #1216
317 lines
13 KiB
Python
317 lines
13 KiB
Python
"""Тесты для backend/app/workers/tasks/scrape_kn.py — Redis-lock owner-token
|
||
и resume_kn_run retry on lock_held (P2 issue #1216).
|
||
|
||
Покрытие:
|
||
- `_region_lock` value содержит уникальный uuid (а не литерал "1");
|
||
- два последовательных вызова получают разные токены (uniqueness);
|
||
- `r.eval` вызывается с Lua check-and-delete: правильный key + наш token;
|
||
- если в Redis под key лежит ЧУЖОЙ token — Lua возвращает 0 (mock-симуляция);
|
||
- `_LOCK_TTL_SECONDS` поднят до 45 мин (полтора max sweep duration);
|
||
- `resume_kn_run` при lock_held → `self.retry(countdown=300, max_retries=12)`,
|
||
а НЕ silent skip (issue #1216 fix #2).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
# ── _region_lock owner-token ──────────────────────────────────────────────────
|
||
|
||
|
||
def test_region_lock_value_is_uuid_token_not_literal_one() -> None:
|
||
"""`r.set` должен получить uuid4-hex (32 hex chars) — НЕ литерал '1'.
|
||
|
||
Это фикс fix #1 из issue #1216: без owner-токена finally безусловно
|
||
делал `r.delete(key)` → можно было снести ЧУЖОЙ lock после TTL-expiry.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = True # acquired
|
||
|
||
with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis):
|
||
with mod._region_lock(66, None) as got_lock:
|
||
assert got_lock is True
|
||
|
||
# `r.set(key, token, nx=True, ex=...)` — token positional[1]
|
||
set_args, _set_kwargs = mock_redis.set.call_args
|
||
token = set_args[1]
|
||
assert token != "1", "lock value must be unique owner-token, not literal '1'"
|
||
assert isinstance(token, str)
|
||
assert len(token) == 32, f"uuid4.hex is 32 chars, got len={len(token)}"
|
||
int(token, 16) # raises if not hex
|
||
|
||
|
||
def test_region_lock_tokens_are_unique_across_calls() -> None:
|
||
"""Два последовательных context-manager call → два РАЗНЫХ токена.
|
||
|
||
Иначе owner-check (Lua GET == ARGV[1]) станет бессмысленным —
|
||
оба worker'а получили бы одинаковый token, и второй снёс бы lock первого.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = True
|
||
|
||
captured_tokens: list[str] = []
|
||
with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis):
|
||
with mod._region_lock(66, None):
|
||
captured_tokens.append(mock_redis.set.call_args[0][1])
|
||
with mod._region_lock(66, None):
|
||
captured_tokens.append(mock_redis.set.call_args[0][1])
|
||
|
||
assert len(captured_tokens) == 2
|
||
assert captured_tokens[0] != captured_tokens[1], "tokens must be unique per call"
|
||
|
||
|
||
def test_region_lock_release_uses_lua_eval_with_own_token() -> None:
|
||
"""В finally вызывается `r.eval(LUA, 1, key, token)` — НЕ `r.delete(key)`.
|
||
|
||
Это фикс fix #1 из issue #1216 (check-and-delete семантика).
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = True
|
||
mock_redis.eval.return_value = 1 # пусть Lua говорит «удалил»
|
||
|
||
with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis):
|
||
with mod._region_lock(66, ["devA"]):
|
||
set_token = mock_redis.set.call_args[0][1]
|
||
|
||
# eval должен быть вызван ровно один раз с правильными аргументами.
|
||
mock_redis.eval.assert_called_once()
|
||
eval_args = mock_redis.eval.call_args[0]
|
||
assert eval_args[0] == mod._RELEASE_LOCK_LUA
|
||
assert eval_args[1] == 1 # numkeys
|
||
assert eval_args[2] == mod._lock_key(66, ["devA"]) # KEYS[1]
|
||
assert eval_args[3] == set_token # ARGV[1] — наш token
|
||
|
||
# delete (без guard) НЕ должен вызываться — был бы регрессией к старому багу.
|
||
mock_redis.delete.assert_not_called()
|
||
|
||
|
||
def test_region_lock_release_skipped_when_not_acquired() -> None:
|
||
"""Если acquire не удался — finally НЕ должен делать eval/delete.
|
||
|
||
Иначе worker, получивший skipped, снёс бы lock владельца.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = None # nx=True + key exists → None
|
||
|
||
with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis):
|
||
with mod._region_lock(66, None) as got_lock:
|
||
assert got_lock is False
|
||
|
||
mock_redis.eval.assert_not_called()
|
||
mock_redis.delete.assert_not_called()
|
||
|
||
|
||
def test_release_lock_lua_script_contract() -> None:
|
||
"""Smoke: Lua script содержит check-and-delete семантику.
|
||
|
||
Если script отредактируют наугад — этот guard упадёт.
|
||
"""
|
||
from app.workers.tasks.scrape_kn import _RELEASE_LOCK_LUA
|
||
|
||
assert 'redis.call("GET", KEYS[1])' in _RELEASE_LOCK_LUA
|
||
assert "ARGV[1]" in _RELEASE_LOCK_LUA
|
||
assert 'redis.call("DEL", KEYS[1])' in _RELEASE_LOCK_LUA
|
||
assert "else return 0 end" in _RELEASE_LOCK_LUA
|
||
|
||
|
||
def test_region_lock_uses_full_ttl_45_min() -> None:
|
||
"""`r.set` получает `ex=_LOCK_TTL_SECONDS == 45*60 == 2700`.
|
||
|
||
Issue #1216 fix: было 30 мин (== max sweep duration → нулевой запас);
|
||
поднимаем до 45 мин = полтора sweep duration.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
assert mod._LOCK_TTL_SECONDS == 45 * 60
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = True
|
||
|
||
with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis):
|
||
with mod._region_lock(66, None):
|
||
pass
|
||
|
||
_, set_kwargs = mock_redis.set.call_args
|
||
assert set_kwargs["ex"] == 2700
|
||
assert set_kwargs["nx"] is True
|
||
|
||
|
||
def test_lua_script_simulation_rejects_foreign_token() -> None:
|
||
"""Симуляция Lua check-and-delete на python-уровне:
|
||
если value под key != нашему token → 0 (не удалили), иначе → 1.
|
||
|
||
Гарантирует что контракт скрипта корректен; реальная атомарность —
|
||
забота Redis (тестируется на integration-уровне).
|
||
"""
|
||
from app.workers.tasks.scrape_kn import _RELEASE_LOCK_LUA
|
||
|
||
def simulate(stored: str | None, our_token: str) -> int:
|
||
# Грубая модель: parse-free assertion на ключевые ветки скрипта.
|
||
assert 'if redis.call("GET", KEYS[1]) == ARGV[1]' in _RELEASE_LOCK_LUA
|
||
assert "else return 0 end" in _RELEASE_LOCK_LUA
|
||
if stored == our_token:
|
||
return 1
|
||
return 0
|
||
|
||
own = "a" * 32
|
||
foreign = "b" * 32
|
||
assert simulate(stored=own, our_token=own) == 1
|
||
assert simulate(stored=foreign, our_token=own) == 0
|
||
assert simulate(stored=None, our_token=own) == 0
|
||
|
||
|
||
# ── resume_kn_run retry on lock_held ──────────────────────────────────────────
|
||
|
||
|
||
def _patch_session_local(mock_db: MagicMock) -> Any:
|
||
"""Helper: SessionLocal импортируется внутри resume_kn_run из `app.core.db`,
|
||
поэтому патчить надо именно там, а не на module-level scrape_kn."""
|
||
return patch("app.core.db.SessionLocal", return_value=mock_db)
|
||
|
||
|
||
def test_resume_kn_run_retries_on_lock_held_not_silent_skip() -> None:
|
||
"""`resume_kn_run` при lock_held → `self.retry(countdown=300, max_retries=12)`.
|
||
|
||
Старое поведение (silent skip → return {'skipped': True}) теряло run до
|
||
недельного beat — это и есть P2 issue #1216 fix #2.
|
||
|
||
Patch'им `mod.resume_kn_run.retry` напрямую: при bind=True Celery передаёт
|
||
task-instance как `self`, и self.retry == task.retry. Подменяем его на
|
||
side_effect-exception, ловим, проверяем kwargs.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
row = {"region_codes": [66], "developer_ids": None}
|
||
mock_result = MagicMock()
|
||
mock_result.mappings.return_value.first.return_value = row
|
||
mock_db = MagicMock()
|
||
mock_db.execute.return_value = mock_result
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = None # lock_held
|
||
|
||
class FakeRetryError(Exception):
|
||
pass
|
||
|
||
with (
|
||
_patch_session_local(mock_db),
|
||
patch.object(mod.redis.Redis, "from_url", return_value=mock_redis),
|
||
patch.object(mod, "_log_task_received"),
|
||
patch.object(mod, "_log_task_skipped"),
|
||
patch.object(
|
||
mod.resume_kn_run, "retry", side_effect=FakeRetryError("re-queue")
|
||
) as mock_retry,
|
||
):
|
||
raised = False
|
||
try:
|
||
mod.resume_kn_run.run(42)
|
||
except FakeRetryError:
|
||
raised = True
|
||
|
||
assert raised, "resume_kn_run must raise (via self.retry), not return"
|
||
mock_retry.assert_called_once()
|
||
retry_kwargs = mock_retry.call_args.kwargs
|
||
assert retry_kwargs["countdown"] == 300
|
||
assert retry_kwargs["max_retries"] == 12
|
||
assert "exc" in retry_kwargs
|
||
|
||
|
||
def test_resume_kn_run_lock_held_does_not_silently_return_skipped() -> None:
|
||
"""Регрессия-guard: при lock_held НЕ возвращаем dict — поднимаем Retry.
|
||
|
||
Подменяем self.retry на sentinel-exception; если код пошёл бы по
|
||
старой ветке (silent skip → return), exception не возник бы.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
row = {"region_codes": [66], "developer_ids": None}
|
||
mock_result = MagicMock()
|
||
mock_result.mappings.return_value.first.return_value = row
|
||
mock_db = MagicMock()
|
||
mock_db.execute.return_value = mock_result
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = None # lock_held
|
||
|
||
class SentinelRetryError(Exception):
|
||
pass
|
||
|
||
result_returned: Any = None
|
||
raised = False
|
||
with (
|
||
_patch_session_local(mock_db),
|
||
patch.object(mod.redis.Redis, "from_url", return_value=mock_redis),
|
||
patch.object(mod, "_log_task_received"),
|
||
patch.object(mod, "_log_task_skipped"),
|
||
patch.object(mod.resume_kn_run, "retry", side_effect=SentinelRetryError()),
|
||
):
|
||
try:
|
||
result_returned = mod.resume_kn_run.run(42)
|
||
except SentinelRetryError:
|
||
raised = True
|
||
|
||
assert raised, "resume_kn_run must call self.retry (raise Retry), not return"
|
||
assert result_returned is None
|
||
|
||
|
||
def test_resume_kn_run_run_not_found_still_returns_skipped() -> None:
|
||
"""Если row не найден — возвращаем `{'skipped': True, 'reason': 'run_not_found'}`.
|
||
|
||
Это легитимный skip (run уже почистили), retry бесполезен. Проверяем что
|
||
fix #2 не сломал валидные ранние выходы.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_result = MagicMock()
|
||
mock_result.mappings.return_value.first.return_value = None
|
||
mock_db = MagicMock()
|
||
mock_db.execute.return_value = mock_result
|
||
|
||
with (
|
||
_patch_session_local(mock_db),
|
||
patch.object(mod.resume_kn_run, "retry") as mock_retry,
|
||
):
|
||
result = mod.resume_kn_run.run(999)
|
||
|
||
assert result == {"skipped": True, "reason": "run_not_found", "run_id": 999}
|
||
mock_retry.assert_not_called()
|
||
|
||
|
||
# ── scrape_kn_region остаётся skipped (а не retry) — это by-design ────────────
|
||
|
||
|
||
def test_scrape_kn_region_lock_held_returns_skipped() -> None:
|
||
"""`scrape_kn_region` (scheduled beat) при lock_held → return skipped.
|
||
|
||
В отличие от resume_kn_run, scheduled run крутится по weekly beat —
|
||
пропустить разовый запуск ок, beat поднимет в следующий раз. Retry'ить
|
||
тут было бы спамом задач.
|
||
"""
|
||
from app.workers.tasks import scrape_kn as mod
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.set.return_value = None # lock_held
|
||
|
||
with (
|
||
patch.object(mod.redis.Redis, "from_url", return_value=mock_redis),
|
||
patch.object(mod, "_log_task_received"),
|
||
patch.object(mod, "_log_task_skipped"),
|
||
patch.object(mod.settings, "scrape_kn_jitter_seconds", 0),
|
||
patch.object(mod.settings, "scrape_kn_state_path", ""),
|
||
patch.object(mod.scrape_kn_region, "retry") as mock_retry,
|
||
):
|
||
result = mod.scrape_kn_region.run(66, None, True)
|
||
|
||
assert result["skipped"] is True
|
||
assert result["reason"] == "lock_held"
|
||
assert result["region_code"] == 66
|
||
mock_retry.assert_not_called()
|