bot-backend
07b06617fd
All checks were successful
CI / changes (push) Successful in 6s
CI / changes (pull_request) Successful in 5s
CI / backend-tests (push) Has been skipped
CI / frontend-tests (push) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""Тесты для scheduler_main.py — standalone scraper-контейнер (#1182).
|
||
|
||
(a) модуль импортируется без реального DB/сети.
|
||
(b) _should_run() возвращает False при SCHEDULER_ENABLE=false.
|
||
(c) _run() отменяется чисто без unhandled exceptions.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import pytest
|
||
|
||
import app.scheduler_main as sm
|
||
|
||
|
||
# (a) Импорт модуля
|
||
def test_scheduler_main_importable() -> None:
|
||
"""scheduler_main импортируется без ошибок."""
|
||
assert sm is not None
|
||
assert hasattr(sm, "_should_run")
|
||
assert hasattr(sm, "_run")
|
||
|
||
|
||
# (b) Kill-switch SCHEDULER_ENABLE=false
|
||
def test_should_run_false_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""_should_run() == False когда SCHEDULER_ENABLE отключён."""
|
||
monkeypatch.setattr(sm.settings, "scheduler_enable", False)
|
||
assert sm._should_run() is False
|
||
|
||
|
||
def test_should_run_true_when_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""_should_run() == True при включённом scheduler."""
|
||
monkeypatch.setattr(sm.settings, "scheduler_enable", True)
|
||
assert sm._should_run() is True
|
||
|
||
|
||
# (c) _run() отменяется чисто
|
||
@pytest.mark.asyncio
|
||
async def test_run_cancels_cleanly(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""_run() завершается без unhandled exceptions при отмене задачи."""
|
||
cancelled = asyncio.Event()
|
||
|
||
async def _stub_loop() -> None:
|
||
"""Короткая заглушка: ждёт отмены, имитируя бесконечный scheduler_loop."""
|
||
cancelled.set()
|
||
await asyncio.sleep(999)
|
||
|
||
# Патчим источник импорта: _run() делает `from app.services.scheduler import scheduler_loop`
|
||
# на каждый вызов, поэтому патч модуля-источника достаточен.
|
||
import app.services.scheduler as _sched_mod
|
||
|
||
monkeypatch.setattr(_sched_mod, "scheduler_loop", _stub_loop)
|
||
|
||
task = asyncio.create_task(sm._run())
|
||
|
||
# Ждём пока _stub_loop запустится (флаг)
|
||
await asyncio.wait_for(cancelled.wait(), timeout=2.0)
|
||
|
||
task.cancel()
|
||
|
||
# Не должно бросить никакого исключения
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
pass # CancelledError от самого task — допустимо
|
||
|
||
# Тест прошёл если не было unhandled exception
|
||
assert task.done()
|