ElebobaandClaude Opus 5.5
282920e06b
- app/db/models.py: 13 таблиц (ARCHITECTURE §2.10); перечисления — VARCHAR + CHECK, JSON → JSONB на Postgres, BIGINT/INTEGER-ключи для SQLite - миграция 0001 (автогенерация + ручная чистка: импорт sa.Text, дубли CHECK) - миграции применяются при старте бота программно (app.db.migrate), script_location из пакета - SqlStorage: FSM aiogram поверх fsm_state — состояние переживает рестарт (перенесено из #4) - cleanup_loop: истёкшие черновики, снимки контейнеров и попытки входа старше суток - тесты: upgrade/downgrade, дрейф моделей и миграций, CHECK, CASCADE, FSM, чистка Refs #5 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
import asyncio
|
|
import contextlib
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import health
|
|
|
|
|
|
def test_missing_file_is_dead(tmp_path: Path) -> None:
|
|
assert not health.is_alive(tmp_path / "hb", max_age=30)
|
|
|
|
|
|
def test_age_decides(tmp_path: Path) -> None:
|
|
hb = tmp_path / "hb"
|
|
hb.touch()
|
|
os.utime(hb, (1000, 1000))
|
|
assert health.is_alive(hb, max_age=30, now=1029)
|
|
assert not health.is_alive(hb, max_age=30, now=1031)
|
|
|
|
|
|
async def test_heartbeat_touches_file(tmp_path: Path) -> None:
|
|
hb = tmp_path / "hb"
|
|
task = asyncio.create_task(health.heartbeat(hb, interval=10))
|
|
async with asyncio.timeout(2):
|
|
while not hb.exists(): # noqa: ASYNC110 - polling the filesystem; touch runs in a worker thread
|
|
await asyncio.sleep(0.01)
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
@pytest.mark.parametrize(("touch", "code"), [(True, 0), (False, 1)])
|
|
def test_cli_exit_code(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, touch: bool, code: int) -> None:
|
|
hb = tmp_path / "hb"
|
|
if touch:
|
|
hb.touch()
|
|
monkeypatch.setenv("HEARTBEAT_FILE", str(hb))
|
|
with pytest.raises(SystemExit) as exc:
|
|
health.main()
|
|
assert exc.value.code == code
|