ElebobaandClaude Opus 5.5
e8c3a321a3
CI / ruff · mypy · pytest (pull_request) Successful in 1h3m54s
- invites.max_uses/uses: обычное приглашение — 1 вход, демо-QR — до DEMO_INVITE_MAX_USES (50), живёт DEMO_INVITE_TTL_HOURS (3 ч); захват атомарный, UPDATE … WHERE uses < max_uses - users.guest: вошедшие по демо-QR; «🧹 Отключить гостей демо» отзывает всех и гасит демо-QR - «👥 Участники»: кнопка «🎤 Демо-QR», гости в списке — 🎟 - миграция 0003 (ADD COLUMN без batch — SQLite не пересоздаёт users); проверена на Postgres - SPEC A-3a, ADR-017, DEMO.md, .env.example Refs #55 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from alembic.autogenerate import compare_metadata
|
|
from alembic.migration import MigrationContext
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
from app.db.migrate import downgrade, upgrade
|
|
from app.db.models import Base, ChangeOp, Draft, DraftChange, User, utcnow
|
|
from app.db.session import SessionMaker, make_engine
|
|
|
|
TABLES = {
|
|
"users",
|
|
"invites",
|
|
"access_attempts",
|
|
"fsm_state",
|
|
"drafts",
|
|
"draft_changes",
|
|
"pushes",
|
|
"pipeline_runs",
|
|
"container_snapshots",
|
|
"alerts",
|
|
"alert_deliveries",
|
|
"log_queries",
|
|
"agent_requests",
|
|
}
|
|
|
|
|
|
async def table_names(engine: AsyncEngine) -> set[str]:
|
|
async with engine.connect() as conn:
|
|
return set(await conn.run_sync(lambda c: inspect(c).get_table_names())) - {"alembic_version"}
|
|
|
|
|
|
async def test_upgrade_creates_all_tables(engine: AsyncEngine) -> None:
|
|
assert await table_names(engine) == TABLES
|
|
|
|
|
|
async def test_models_match_migrations(engine: AsyncEngine) -> None:
|
|
# A model change without a migration shows up here as a non-empty diff.
|
|
async with engine.connect() as conn:
|
|
diff = await conn.run_sync(lambda c: compare_metadata(MigrationContext.configure(c), Base.metadata))
|
|
assert diff == []
|
|
|
|
|
|
async def test_downgrade_to_base_and_back() -> None:
|
|
eng = make_engine("sqlite+aiosqlite://")
|
|
try:
|
|
await upgrade(eng)
|
|
await downgrade(eng, "base")
|
|
assert await table_names(eng) == set()
|
|
await upgrade(eng)
|
|
assert await table_names(eng) == TABLES
|
|
finally:
|
|
await eng.dispose()
|
|
|
|
|
|
INSERT_USER = text(
|
|
"INSERT INTO users (tg_id, full_name, role, is_active, joined_at) VALUES (:id, 'u', :role, 1, '2026-01-01')"
|
|
)
|
|
|
|
|
|
async def test_enum_check_rejects_garbage(engine: AsyncEngine) -> None:
|
|
async with engine.begin() as conn:
|
|
await conn.execute(INSERT_USER, {"id": 1, "role": "admin"})
|
|
with pytest.raises(IntegrityError):
|
|
async with engine.begin() as conn:
|
|
await conn.execute(INSERT_USER, {"id": 2, "role": "root"})
|
|
|
|
|
|
async def test_deleting_draft_cascades_to_changes(sessions: SessionMaker) -> None:
|
|
async with sessions.begin() as s:
|
|
s.add(User(tg_id=1, full_name="a"))
|
|
await s.flush()
|
|
draft = Draft(user_id=1, base_sha="0" * 40, expires_at=utcnow() + timedelta(hours=1))
|
|
draft.changes.append(DraftChange(path="services/a/main.py", op=ChangeOp.ADD, content=b"print(1)"))
|
|
s.add(draft)
|
|
async with sessions.begin() as s:
|
|
await s.execute(text("DELETE FROM drafts"))
|
|
async with sessions() as s:
|
|
assert (await s.execute(text("SELECT count(*) FROM draft_changes"))).scalar_one() == 0
|
|
|
|
|
|
async def test_0003_marks_redeemed_invites_spent() -> None:
|
|
eng = make_engine("sqlite+aiosqlite://")
|
|
try:
|
|
await upgrade(eng, "0002")
|
|
async with eng.begin() as conn:
|
|
await conn.execute(INSERT_USER, {"id": 1, "role": "admin"})
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO invites (token_hash, created_by, created_at, expires_at, used_by) VALUES "
|
|
"('a', 1, '2026-01-01', '2026-01-02', 1), ('b', 1, '2026-01-01', '2026-01-02', NULL)"
|
|
)
|
|
)
|
|
await upgrade(eng)
|
|
async with eng.connect() as conn:
|
|
rows = (await conn.execute(text("SELECT token_hash, max_uses, uses FROM invites ORDER BY 1"))).all()
|
|
assert [tuple(r) for r in rows] == [("a", 1, 1), ("b", 1, 0)]
|
|
assert (await conn.execute(text("SELECT guest FROM users"))).scalar_one() in (0, False)
|
|
finally:
|
|
await eng.dispose()
|