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>
116 lines
4.6 KiB
Python
116 lines
4.6 KiB
Python
import re
|
|
|
|
import pytest
|
|
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage, SendPhoto
|
|
from aiogram.types import ReplyKeyboardMarkup
|
|
|
|
from app.access.service import AccessService
|
|
from app.bot.routers.members import MembersAction, MembersCb
|
|
from app.db.session import SessionMaker
|
|
from app.main import build_dispatcher
|
|
from app.repo.git import UploadsRepo
|
|
from app.texts import Texts
|
|
from tests.conftest import ADMIN
|
|
from tests.fakes import TgClient
|
|
|
|
NEWBIE = 3003
|
|
|
|
|
|
@pytest.fixture
|
|
def tg(texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo) -> TgClient:
|
|
return TgClient(build_dispatcher(texts, sessions, access, repo))
|
|
|
|
|
|
async def invite_token(tg: TgClient, action: MembersAction = MembersAction.INVITE) -> str:
|
|
await tg.press(ADMIN, MembersCb(action=action).pack())
|
|
[photo] = tg.session.sent(SendPhoto)
|
|
match = re.search(r"https://t\.me/tgit_test_bot\?start=([A-Za-z0-9_-]+)", photo.caption or "")
|
|
assert match, photo.caption
|
|
tg.session.requests.clear()
|
|
return match.group(1)
|
|
|
|
|
|
async def test_admin_sees_members_list(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(ADMIN, texts.menu.members)
|
|
|
|
[reply] = tg.session.sent(SendMessage)
|
|
assert reply.text.startswith(texts.members.title.format(count=1))
|
|
|
|
|
|
async def test_full_invite_flow(tg: TgClient, texts: Texts) -> None:
|
|
token = await invite_token(tg)
|
|
|
|
await tg.send(NEWBIE, f"/start {token}")
|
|
|
|
[welcome] = tg.session.sent(SendMessage)
|
|
assert welcome.text == texts.start.welcome
|
|
assert isinstance(welcome.reply_markup, ReplyKeyboardMarkup)
|
|
labels = [b.text for row in welcome.reply_markup.keyboard for b in row]
|
|
assert texts.menu.members not in labels # a member, not an admin
|
|
|
|
# the same QR again, from someone else
|
|
tg.session.requests.clear()
|
|
await tg.send(NEWBIE + 1, f"/start {token}")
|
|
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.invite_rejected]
|
|
|
|
|
|
async def test_bad_token_rejected_and_bare_start_refused(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(NEWBIE, "/start nope")
|
|
await tg.send(NEWBIE + 1, "/start")
|
|
|
|
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.invite_rejected, texts.start.no_access]
|
|
|
|
|
|
async def test_revoke_with_confirmation_cuts_access(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(NEWBIE, f"/start {await invite_token(tg)}")
|
|
tg.session.requests.clear()
|
|
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.ASK_REVOKE, tg_id=NEWBIE).pack())
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.REVOKE, tg_id=NEWBIE).pack())
|
|
|
|
confirm, done = tg.session.sent(EditMessageText)
|
|
assert confirm.text == texts.members.confirm.format(name=f"u{NEWBIE}")
|
|
assert done.text.startswith(texts.members.revoked.format(name=f"u{NEWBIE}"))
|
|
|
|
tg.session.requests.clear()
|
|
await tg.send(NEWBIE, texts.menu.commit)
|
|
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.no_access]
|
|
|
|
|
|
async def test_member_cannot_use_admin_actions(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(NEWBIE, f"/start {await invite_token(tg)}")
|
|
tg.session.requests.clear()
|
|
|
|
await tg.send(NEWBIE, texts.menu.members)
|
|
await tg.press(NEWBIE, MembersCb(action=MembersAction.INVITE).pack())
|
|
|
|
assert tg.session.sent(SendMessage) == [] and tg.session.sent(SendPhoto) == []
|
|
assert len(tg.session.sent(AnswerCallbackQuery)) == 1
|
|
|
|
|
|
async def test_cancel_revoke_returns_to_list(tg: TgClient, texts: Texts) -> None:
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.LIST).pack())
|
|
|
|
[edit] = tg.session.sent(EditMessageText)
|
|
assert edit.text.startswith(texts.members.title.format(count=1))
|
|
|
|
|
|
async def test_demo_qr_lets_many_in_and_guests_are_revoked_together(tg: TgClient, texts: Texts) -> None:
|
|
token = await invite_token(tg, MembersAction.DEMO_INVITE)
|
|
|
|
for n in range(3):
|
|
await tg.send(NEWBIE + n, f"/start {token}")
|
|
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.welcome] * 3
|
|
|
|
tg.session.requests.clear()
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.ASK_REVOKE_GUESTS).pack())
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.REVOKE_GUESTS).pack())
|
|
confirm, done = tg.session.sent(EditMessageText)
|
|
assert confirm.text == texts.members.confirm_guests
|
|
assert done.text.startswith(texts.members.guests_revoked.format(count=3))
|
|
|
|
tg.session.requests.clear()
|
|
await tg.send(NEWBIE, texts.menu.commit)
|
|
await tg.send(NEWBIE + 10, f"/start {token}")
|
|
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.no_access, texts.start.invite_rejected]
|