- 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>
31 lines
1.4 KiB
Python
31 lines
1.4 KiB
Python
from datetime import timedelta
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.db.maintenance import cleanup_expired
|
|
from app.db.models import AccessAttempt, ContainerSnapshot, Draft, User, utcnow
|
|
from app.db.session import SessionMaker
|
|
|
|
|
|
async def test_cleanup_removes_only_expired(sessions: SessionMaker) -> None:
|
|
now = utcnow()
|
|
async with sessions.begin() as s:
|
|
s.add_all([User(tg_id=1, full_name="a"), User(tg_id=2, full_name="b")])
|
|
await s.flush() # no relationship() to order inserts: users first for the FK
|
|
s.add_all(
|
|
[
|
|
Draft(user_id=1, base_sha="a" * 40, expires_at=now - timedelta(minutes=1)),
|
|
Draft(user_id=2, base_sha="b" * 40, expires_at=now + timedelta(hours=1)),
|
|
ContainerSnapshot(container="c", at=now - timedelta(days=2), restart_count=0, cpu_pct=0, mem_pct=0),
|
|
ContainerSnapshot(container="c", at=now, restart_count=0, cpu_pct=0, mem_pct=0),
|
|
AccessAttempt(tg_id=9, at=now - timedelta(days=2), ok=False),
|
|
]
|
|
)
|
|
|
|
removed = await cleanup_expired(sessions, now)
|
|
|
|
assert removed == {"drafts": 1, "snapshots": 1, "access_attempts": 1}
|
|
async with sessions() as s:
|
|
assert await s.scalar(select(func.count()).select_from(Draft)) == 1
|
|
assert await s.scalar(select(func.count()).select_from(ContainerSnapshot)) == 1
|