ElebobaandClaude Opus 5.5
f89338e015
CI / ruff · mypy · pytest (pull_request) Successful in 9m40s
- app/logs/grep.py: подмножество grep (шаблон, -i -v -E -c -C N), разбор shlex без исполнения, BRE → regex, таймаут на строку и бюджет на поиск, лимиты 200 символов / 500 строк вывода - app/logs/journal.py: каждый поиск в log_queries (L-6) - бот: выбор контейнера, хвост 50/200 строк или 5 мин/1 ч/24 ч, свой grep по 24 ч, длинный результат — файлом .log (L-5) - логи Docker: не больше 50 000 строк и 5 МБ, сохраняются новейшие - fix: кнопки меню и /menu, /help больше не проглатываются шагами ввода (имя файла, содержимое, сообщение коммита, grep) — фильтр free_input Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
175 lines
6.8 KiB
Python
175 lines
6.8 KiB
Python
import pytest
|
|
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage
|
|
from aiogram.types import InlineKeyboardMarkup
|
|
|
|
from app.access.service import AccessService
|
|
from app.bot.routers.commit import Act, CommitCb
|
|
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 cb(action: Act, i: int = 0) -> str:
|
|
return CommitCb(a=action, i=i).pack()
|
|
|
|
|
|
def labels(markup: object) -> list[str]:
|
|
assert isinstance(markup, InlineKeyboardMarkup)
|
|
return [b.text for row in markup.inline_keyboard for b in row]
|
|
|
|
|
|
def last_text(tg: TgClient) -> SendMessage:
|
|
return tg.session.sent(SendMessage)[-1]
|
|
|
|
|
|
async def to_hello(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
|
|
|
|
|
|
async def test_new_file_from_fenced_text(tg: TgClient, texts: Texts) -> None:
|
|
await to_hello(tg, texts)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, "utils.py")
|
|
await tg.send(ADMIN, "```python\ndef add(a, b):\n return a + b\n```")
|
|
|
|
reply = last_text(tg)
|
|
assert reply.text.startswith(texts.commit.added.format(path="services/hello/utils.py"))
|
|
got = labels(reply.reply_markup)
|
|
assert "🆕 📄 utils.py · 31 Б" in got # fence stripped, no trailing newline
|
|
assert texts.commit.draft_button.format(count=1) in got
|
|
|
|
await tg.press(ADMIN, cb(Act.OPEN, 1)) # utils.py (main.py is 0)
|
|
await tg.press(ADMIN, cb(Act.VIEW))
|
|
view = last_text(tg)
|
|
assert "def add(a, b):" in view.text and "```" not in view.text
|
|
|
|
|
|
async def test_replace_with_document_shows_diff(tg: TgClient, texts: Texts) -> None:
|
|
await to_hello(tg, texts)
|
|
await tg.press(ADMIN, cb(Act.OPEN, 0)) # main.py
|
|
await tg.press(ADMIN, cb(Act.REPLACE))
|
|
await tg.send_document(ADMIN, "main.py", b"print('hello')\nprint('bye')\n")
|
|
|
|
reply = last_text(tg)
|
|
assert reply.text.startswith(texts.commit.modified.format(path="services/hello/main.py", added=1, removed=0))
|
|
assert "✏️ 📄 main.py · 28 Б" in labels(reply.reply_markup)
|
|
|
|
|
|
async def test_delete_marks_and_draft_lists(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(ADMIN, texts.menu.commit)
|
|
await tg.press(ADMIN, cb(Act.OPEN, 3)) # README.md
|
|
await tg.press(ADMIN, cb(Act.DELETE))
|
|
deleted = tg.session.sent(EditMessageText)[-1]
|
|
assert deleted.text.startswith(texts.commit.deleted.format(path="README.md"))
|
|
assert "🗑 📄 README.md · 10 Б" in labels(deleted.reply_markup)
|
|
|
|
await tg.press(ADMIN, cb(Act.DRAFT))
|
|
draft = tg.session.sent(EditMessageText)[-1]
|
|
assert "D /README.md" in draft.text
|
|
|
|
await tg.press(ADMIN, cb(Act.UNSTAGE, 0))
|
|
assert tg.session.sent(EditMessageText)[-1].text == texts.commit.draft_empty
|
|
|
|
|
|
async def test_syntax_error_force_or_fix(tg: TgClient, texts: Texts) -> None:
|
|
await to_hello(tg, texts)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, "bad.py")
|
|
await tg.send(ADMIN, "def f(:\n pass")
|
|
|
|
warn = last_text(tg)
|
|
assert warn.text.startswith("⚠️") and "1" in warn.text
|
|
assert labels(warn.reply_markup) == [texts.commit.fix, texts.commit.force]
|
|
|
|
await tg.press(ADMIN, cb(Act.FIX))
|
|
assert tg.session.sent(EditMessageText)[-1].text == texts.commit.ask_content.format(path="services/hello/bad.py")
|
|
await tg.send(ADMIN, "def f(:\n pass")
|
|
await tg.press(ADMIN, cb(Act.FORCE))
|
|
assert last_text(tg).text.startswith(texts.commit.added.format(path="services/hello/bad.py"))
|
|
|
|
|
|
async def test_virtual_folder_becomes_real_with_a_file(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.NEW_DIR))
|
|
await tg.send(ADMIN, "api")
|
|
inside = last_text(tg)
|
|
assert inside.text == f"{texts.commit.title.format(path='services/api')}\n{texts.commit.empty}"
|
|
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, "main.py")
|
|
await tg.send(ADMIN, "print('api')")
|
|
await tg.press(ADMIN, cb(Act.UP))
|
|
|
|
services = tg.session.sent(EditMessageText)[-1]
|
|
assert "🆕 📁 api" in labels(services.reply_markup)
|
|
|
|
|
|
async def test_names_rejected(tg: TgClient, texts: Texts) -> None:
|
|
await tg.send(ADMIN, texts.menu.commit)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
for bad in ["../etc", "README.md", "ruff.toml"]:
|
|
await tg.send(ADMIN, bad)
|
|
|
|
assert [m.text for m in tg.session.sent(SendMessage)][-3:] == [
|
|
texts.commit.bad_name,
|
|
texts.commit.exists,
|
|
texts.commit.protected,
|
|
]
|
|
|
|
|
|
async def test_binary_document_rejected(tg: TgClient, texts: Texts) -> None:
|
|
await to_hello(tg, texts)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, "img.png")
|
|
await tg.send_document(ADMIN, "img.png", b"\x89PNG\x00\x00")
|
|
|
|
assert last_text(tg).text == texts.commit.binary
|
|
|
|
|
|
async def test_draft_survives_a_restart(
|
|
texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo
|
|
) -> None:
|
|
first = TgClient(build_dispatcher(texts, sessions, access, repo))
|
|
await first.send(ADMIN, texts.menu.commit)
|
|
await first.press(ADMIN, cb(Act.OPEN, 3))
|
|
await first.press(ADMIN, cb(Act.DELETE))
|
|
|
|
second = TgClient(build_dispatcher(texts, sessions, access, repo)) # same DB, new bot process
|
|
await second.send(ADMIN, texts.menu.commit)
|
|
assert texts.commit.draft_button.format(count=1) in labels(second.session.sent(SendMessage)[-1].reply_markup)
|
|
|
|
|
|
async def test_protected_file_cannot_be_replaced(tg: TgClient, texts: Texts, repo: UploadsRepo) -> None:
|
|
await tg.send(ADMIN, texts.menu.commit)
|
|
# fake a card for a protected path via FSM: open README then rewrite the selection
|
|
await tg.press(ADMIN, cb(Act.OPEN, 3))
|
|
state = tg.dp.fsm.get_context(tg.bot, chat_id=ADMIN, user_id=ADMIN)
|
|
await state.update_data(commit_file=".sourcecraft/ci.yaml")
|
|
await tg.press(ADMIN, cb(Act.REPLACE))
|
|
|
|
answer = tg.session.sent(AnswerCallbackQuery)[-1]
|
|
assert answer.text == texts.commit.protected and answer.show_alert
|
|
|
|
|
|
@pytest.mark.parametrize("escape", ["menu_status", "/menu", "/help"])
|
|
async def test_menu_and_commands_escape_input_states(tg: TgClient, texts: Texts, escape: str) -> None:
|
|
text = texts.menu.status if escape == "menu_status" else escape
|
|
await to_hello(tg, texts)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, text)
|
|
|
|
reply = last_text(tg)
|
|
assert reply.text != texts.commit.bad_name
|
|
assert not reply.text.startswith(texts.commit.ask_content.split("{")[0])
|