ElebobaandClaude Opus 5.5
219bebd5a2
- AccessService: одноразовые приглашения (128 бит, в БД только sha256, TTL 24 ч), атомарный захват приглашения (условный UPDATE — из двух сканов одного QR проходит один), антиперебор (> 5 неудач в час — молчание), отзыв (черновик и FSM удаляются), повторное приглашение - ADMIN_IDS заводятся в БД при старте; дальше — только по приглашению - AccessMiddleware: участники из БД (+ last_seen раз в 5 мин), новичкам — только /start <токен> - «👥 Участники» для админов: список, QR (segno) + ссылка, отзыв с подтверждением - один ответ на неверный/истёкший/использованный токен — без подсказок Refs #25 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
95 lines
3.5 KiB
Python
95 lines
3.5 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.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) -> TgClient:
|
|
return TgClient(build_dispatcher(texts, sessions, access))
|
|
|
|
|
|
async def invite_token(tg: TgClient) -> str:
|
|
await tg.press(ADMIN, MembersCb(action=MembersAction.INVITE).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))
|