ElebobaandClaude Opus 5.5
ae0cd5d42d
- Publisher: пуши по одному (lock), fetch → проверка конфликта с base_sha → чистое дерево origin/main → черновик → коммит от tgit-bot с трейлером Telegram-User → push; отказ push → rebase и повтор (до 3 раз); защита от путей вне клона; лимит 20 пушей/час - «💬 Сообщение коммита» (1–200 символов, первая строка ≤ 72) и «🚀 Push» с подтверждением; исходы: запушено (sha, ссылка, ⏳ CI) · конфликт → «Перезаписать» · нечего коммитить · лимит · сбой - запись в pushes (chat_id, message_id — для обновления статуса CI/CD в #11), черновик очищается - ruff: RUF001–003 выключены — кириллица в текстах это норма Refs #9 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
66 lines
2.7 KiB
Python
66 lines
2.7 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from aiogram.methods import 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, git
|
|
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, "https://example.test/commit/{sha}"))
|
|
|
|
|
|
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]
|
|
|
|
|
|
async def test_message_then_push_lands_in_remote(tg: TgClient, texts: Texts, remote: Path, tmp_path: Path) -> None:
|
|
t = texts.commit
|
|
await tg.send(ADMIN, texts.menu.commit)
|
|
await tg.press(ADMIN, cb(Act.NEW_FILE))
|
|
await tg.send(ADMIN, "notes.txt")
|
|
await tg.send(ADMIN, "привет из телеграма")
|
|
await tg.press(ADMIN, cb(Act.DRAFT))
|
|
draft = tg.session.sent(EditMessageText)[-1]
|
|
assert t.message_missing in draft.text and t.push_button not in labels(draft.reply_markup)
|
|
|
|
await tg.press(ADMIN, cb(Act.MESSAGE))
|
|
await tg.send(ADMIN, "x" * 80) # first line too long
|
|
assert tg.session.sent(SendMessage)[-1].text == t.bad_message
|
|
await tg.send(ADMIN, "docs: заметки")
|
|
with_message = tg.session.sent(SendMessage)[-1]
|
|
assert t.message_current.format(message="docs: заметки") in with_message.text
|
|
assert t.push_button in labels(with_message.reply_markup)
|
|
|
|
await tg.press(ADMIN, cb(Act.PUSH))
|
|
assert tg.session.sent(EditMessageText)[-1].text.startswith("Запушить в main?")
|
|
await tg.press(ADMIN, cb(Act.PUSH_GO))
|
|
|
|
pushing, done = tg.session.sent(EditMessageText)[-2:]
|
|
assert pushing.text == t.pushing
|
|
assert done.text.startswith("✅ Запушено: ") and "docs: заметки" in done.text
|
|
assert "https://example.test/commit/" in done.text and done.text.endswith(t.ci_pending)
|
|
|
|
clone = tmp_path / "verify"
|
|
git(tmp_path, "clone", "-q", str(remote), str(clone))
|
|
assert (clone / "notes.txt").read_text(encoding="utf-8") == "привет из телеграма"
|
|
|
|
# the draft is gone after a push
|
|
tg.session.requests.clear()
|
|
await tg.send(ADMIN, texts.menu.commit)
|
|
assert not any(label.startswith("🧾") for label in labels(tg.session.sent(SendMessage)[-1].reply_markup))
|