import pytest from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendDocument, SendMessage from aiogram.types import InlineKeyboardMarkup from app.access.service import AccessService from app.bot.routers.commit import Act, CommitCb, human_size 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 @pytest.fixture def tg(texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo) -> TgClient: return TgClient(build_dispatcher(texts, sessions, access, repo)) def labels(markup: object) -> list[str]: assert isinstance(markup, InlineKeyboardMarkup) return [b.text for row in markup.inline_keyboard for b in row] def cb(action: Act, i: int = 0) -> str: return CommitCb(a=action, i=i).pack() async def test_root_listing(tg: TgClient, texts: Texts) -> None: await tg.send(ADMIN, texts.menu.commit) [reply] = tg.session.sent(SendMessage) assert reply.text == texts.commit.title.format(path="") got = labels(reply.reply_markup) assert got[:4] == ["📁 docs", "📁 many", "📁 services", "📄 README.md · 10 Б"] assert texts.commit.up not in got # already at the root async def test_open_dir_then_file_card_then_back(tg: TgClient, texts: Texts) -> None: await tg.send(ADMIN, texts.menu.commit) await tg.press(ADMIN, cb(Act.OPEN, 2)) # services await tg.press(ADMIN, cb(Act.OPEN, 0)) # services/hello await tg.press(ADMIN, cb(Act.OPEN, 0)) # services/hello/main.py hello_dir, hello_sub, card = tg.session.sent(EditMessageText) assert hello_dir.text == texts.commit.title.format(path="services") assert texts.commit.up in labels(hello_dir.reply_markup) assert hello_sub.text == texts.commit.title.format(path="services/hello") assert card.text.startswith("📄 /services/hello/main.py") and "init" in card.text assert labels(card.reply_markup) == [ texts.commit.view, texts.commit.replace, texts.commit.delete, texts.commit.to_dir, ] await tg.press(ADMIN, cb(Act.TO_DIR)) await tg.press(ADMIN, cb(Act.UP)) *_, back_to_dir, up = tg.session.sent(EditMessageText) assert back_to_dir.text == texts.commit.title.format(path="services/hello") assert up.text == texts.commit.title.format(path="services") async def test_view_text_inline_and_binary_as_document(tg: TgClient, texts: Texts) -> None: await tg.send(ADMIN, texts.menu.commit) await tg.press(ADMIN, cb(Act.OPEN, 0)) # docs await tg.press(ADMIN, cb(Act.OPEN, 1)) # мои заметки.md await tg.press(ADMIN, cb(Act.VIEW)) await tg.press(ADMIN, cb(Act.TO_DIR)) await tg.press(ADMIN, cb(Act.OPEN, 0)) # logo.bin await tg.press(ADMIN, cb(Act.VIEW)) inline = [m for m in tg.session.sent(SendMessage) if m.parse_mode == "HTML"] assert len(inline) == 1 and "
привет\n" in inline[0].text [doc] = tg.session.sent(SendDocument) assert doc.caption == texts.commit.sent_as_file async def test_pagination(tg: TgClient, texts: Texts) -> None: await tg.send(ADMIN, texts.menu.commit) await tg.press(ADMIN, cb(Act.OPEN, 1)) # many: 25 files await tg.press(ADMIN, cb(Act.PAGE, 1)) first, second = tg.session.sent(EditMessageText) assert texts.commit.page.format(page=1, pages=2) in first.text assert labels(first.reply_markup)[0].startswith("📄 f00.txt") and texts.commit.next in labels(first.reply_markup) assert labels(second.reply_markup)[0].startswith("📄 f20.txt") and texts.commit.prev in labels(second.reply_markup) async def test_unfinished_actions_and_stale_buttons(tg: TgClient, texts: Texts) -> None: await tg.send(ADMIN, texts.menu.commit) await tg.press(ADMIN, cb(Act.NEW_FILE)) await tg.press(ADMIN, cb(Act.OPEN, 99)) soon, stale = tg.session.sent(AnswerCallbackQuery) assert soon.text == texts.commit.soon and soon.show_alert assert stale.text is None assert tg.session.sent(EditMessageText) == [] @pytest.mark.parametrize(("n", "text"), [(10, "10 Б"), (2048, "2.0 КБ"), (5 * 1024 * 1024, "5.0 МБ")]) def test_human_size(n: int, text: str) -> None: assert human_size(n) == text