ElebobaandClaude Opus 5.5
dbb17bd774
- UploadsRepo: локальный клон, fetch не чаще fetch_interval, ls-tree -z (пробелы/кириллица), cat-file, последний коммит по пути; git без shell, ssh с -F /dev/null (без конфига хоста) - «📝 Коммит»: папки/файлы с размерами, ⬆️ Вверх, страницы по 20, карточка файла с последним коммитом, «👁 Показать» (<pre> или документом); в кнопках только индексы - ключ для контейнера: GIT_SSH_KEY_B64 → временный файл 0600 (смонтированный файл хоста контейнерному uid не прочитать); volume /data для клона - тесты на настоящем bare-репозитории во временной папке Refs #8 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
96 lines
3.6 KiB
Python
96 lines
3.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) -> 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))
|