feat(bot): Коммит — создание, замена и удаление файлов, черновик
- новый файл: имя → содержимое текстом (```-обёртка снимается) или документом - новая папка (виртуальная, пока в ней нет файлов), замена с diff +N/−M, удаление - валидация: имя, ≤ 100 КБ, UTF-8, без NUL; .py — ast.parse с выбором «исправить / всё равно»; .sourcecraft/, .tgit/, ruff.toml из бота не меняются - DraftService: черновик в БД, 24 ч, ≤ 50 изменений; удаление нового файла убирает его из черновика - дерево показывает черновик поверх репозитория: 🆕 / ✏️ / 🗑, размер — как будет после пуша - экран черновика (A/M/D) с «↩️ убрать» Refs #26 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -87,13 +87,11 @@ async def test_pagination(tg: TgClient, texts: Texts) -> None:
|
||||
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:
|
||||
async def test_stale_button_is_ignored(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
|
||||
[stale] = tg.session.sent(AnswerCallbackQuery)
|
||||
assert stale.text is None
|
||||
assert tg.session.sent(EditMessageText) == []
|
||||
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
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
|
||||
+22
-5
@@ -7,9 +7,9 @@ from typing import Any
|
||||
|
||||
from aiogram import Bot, Dispatcher
|
||||
from aiogram.client.session.base import BaseSession
|
||||
from aiogram.methods import EditMessageText, GetMe, SendMessage, SendPhoto, TelegramMethod
|
||||
from aiogram.methods import EditMessageText, GetFile, GetMe, SendMessage, SendPhoto, TelegramMethod
|
||||
from aiogram.methods.base import TelegramType
|
||||
from aiogram.types import Chat, Message, Update, User
|
||||
from aiogram.types import Chat, File, Message, Update, User
|
||||
|
||||
BOT_TOKEN = "42:TEST-TOKEN"
|
||||
|
||||
@@ -19,6 +19,7 @@ class FakeSession(BaseSession):
|
||||
super().__init__()
|
||||
self.requests: list[TelegramMethod[Any]] = []
|
||||
self._ids = count(1000)
|
||||
self.files: dict[str, bytes] = {} # file_id -> content, for documents users "upload"
|
||||
|
||||
async def make_request(
|
||||
self,
|
||||
@@ -27,6 +28,10 @@ class FakeSession(BaseSession):
|
||||
timeout: int | None = None, # noqa: ASYNC109 - signature fixed by aiogram BaseSession
|
||||
) -> TelegramType:
|
||||
self.requests.append(method)
|
||||
if isinstance(method, GetFile):
|
||||
return File( # type: ignore[return-value]
|
||||
file_id=method.file_id, file_unique_id=method.file_id, file_path=f"documents/{method.file_id}"
|
||||
)
|
||||
if isinstance(method, GetMe):
|
||||
return User(id=42, is_bot=True, first_name="tgit", username="tgit_test_bot") # type: ignore[return-value]
|
||||
if isinstance(method, SendPhoto):
|
||||
@@ -54,9 +59,9 @@ class FakeSession(BaseSession):
|
||||
timeout: int = 30, # noqa: ASYNC109 - signature fixed by aiogram BaseSession
|
||||
chunk_size: int = 65536,
|
||||
raise_for_status: bool = True,
|
||||
) -> AsyncGenerator[bytes, None]: # pragma: no cover - not used
|
||||
raise NotImplementedError
|
||||
yield b""
|
||||
) -> AsyncGenerator[bytes, None]:
|
||||
file_id = url.rsplit("/", 1)[-1]
|
||||
yield self.files[file_id]
|
||||
|
||||
async def close(self) -> None:
|
||||
pass
|
||||
@@ -92,6 +97,18 @@ class TgClient:
|
||||
message["entities"] = [{"type": "bot_command", "offset": 0, "length": len(cmd)}]
|
||||
await self._feed({"message": message})
|
||||
|
||||
async def send_document(self, tg_id: int, name: str, content: bytes) -> None:
|
||||
file_id = f"doc{next(self._message_ids)}"
|
||||
self.session.files[file_id] = content
|
||||
message = {
|
||||
"message_id": next(self._message_ids),
|
||||
"date": int(datetime.now(UTC).timestamp()),
|
||||
"chat": {"id": tg_id, "type": "private"},
|
||||
"from": self.user(tg_id),
|
||||
"document": {"file_id": file_id, "file_unique_id": file_id, "file_name": name, "file_size": len(content)},
|
||||
}
|
||||
await self._feed({"message": message})
|
||||
|
||||
async def press(self, tg_id: int, data: str, message_text: str = "section") -> None:
|
||||
callback = {
|
||||
"id": str(next(self._update_ids)),
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from app.db.models import ChangeOp, User
|
||||
from app.db.session import SessionMaker
|
||||
from app.repo.drafts import MAX_CHANGES, DraftFull, DraftService
|
||||
|
||||
UID = 7
|
||||
SHA = "a" * 40
|
||||
|
||||
|
||||
class Clock:
|
||||
now = datetime(2026, 10, 15, tzinfo=UTC)
|
||||
|
||||
def __call__(self) -> datetime:
|
||||
return self.now
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def svc(sessions: SessionMaker) -> DraftService:
|
||||
async with sessions.begin() as s:
|
||||
s.add(User(tg_id=UID, full_name="u"))
|
||||
return DraftService(sessions, clock=Clock())
|
||||
|
||||
|
||||
async def test_put_get_and_order(svc: DraftService) -> None:
|
||||
await svc.put(UID, SHA, "b.py", ChangeOp.ADD, b"b")
|
||||
await svc.put(UID, "other-sha", "a.py", ChangeOp.MODIFY, b"a")
|
||||
|
||||
draft = await svc.get(UID)
|
||||
assert draft is not None and draft.base_sha == SHA # base fixed when the draft started
|
||||
assert [(c.path, c.op) for c in draft.changes] == [("b.py", ChangeOp.ADD), ("a.py", ChangeOp.MODIFY)]
|
||||
|
||||
|
||||
async def test_editing_a_new_file_keeps_it_added_and_deleting_it_drops_it(svc: DraftService) -> None:
|
||||
await svc.put(UID, SHA, "n.py", ChangeOp.ADD, b"1")
|
||||
await svc.put(UID, SHA, "n.py", ChangeOp.MODIFY, b"2")
|
||||
draft = await svc.get(UID)
|
||||
assert draft is not None and draft.changes[0].op is ChangeOp.ADD and draft.changes[0].content == b"2"
|
||||
|
||||
await svc.put(UID, SHA, "n.py", ChangeOp.DELETE, None)
|
||||
draft = await svc.get(UID)
|
||||
assert draft is not None and draft.changes == []
|
||||
|
||||
|
||||
async def test_drop_message_clear(svc: DraftService) -> None:
|
||||
await svc.put(UID, SHA, "a.py", ChangeOp.ADD, b"1")
|
||||
await svc.put(UID, SHA, "b.py", ChangeOp.DELETE, None)
|
||||
await svc.drop(UID, "a.py")
|
||||
await svc.set_message(UID, "feat: x")
|
||||
|
||||
draft = await svc.get(UID)
|
||||
assert draft is not None and [c.path for c in draft.changes] == ["b.py"] and draft.message == "feat: x"
|
||||
await svc.clear(UID)
|
||||
assert await svc.get(UID) is None
|
||||
|
||||
|
||||
async def test_expired_draft_is_gone(svc: DraftService) -> None:
|
||||
await svc.put(UID, SHA, "a.py", ChangeOp.ADD, b"1")
|
||||
Clock.now += timedelta(hours=25)
|
||||
try:
|
||||
assert await svc.get(UID) is None
|
||||
finally:
|
||||
Clock.now -= timedelta(hours=25)
|
||||
|
||||
|
||||
async def test_draft_size_limit(svc: DraftService) -> None:
|
||||
for i in range(MAX_CHANGES):
|
||||
await svc.put(UID, SHA, f"f{i}.py", ChangeOp.ADD, b"")
|
||||
with pytest.raises(DraftFull):
|
||||
await svc.put(UID, SHA, "one-more.py", ChangeOp.ADD, b"")
|
||||
@@ -0,0 +1,74 @@
|
||||
import pytest
|
||||
|
||||
from app.repo.validate import (
|
||||
ValidationError,
|
||||
check_content,
|
||||
check_name,
|
||||
diff_stat,
|
||||
is_protected,
|
||||
join,
|
||||
python_syntax,
|
||||
strip_fence,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["main.py", "a-b_c.1", ".gitignore", "X" * 100])
|
||||
def test_good_names(name: str) -> None:
|
||||
assert check_name(f" {name} ") == name
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["", ".", "..", "a/b", "привет.py", "a b", "X" * 101, "a;rm"])
|
||||
def test_bad_names(name: str) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
check_name(name)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("path", "protected"),
|
||||
[
|
||||
(".sourcecraft/ci.yaml", True),
|
||||
(".tgit/x.py", True),
|
||||
("ruff.toml", True),
|
||||
(".sourcecraft", True),
|
||||
("services/a/ruff.toml", False),
|
||||
("README.md", False),
|
||||
("services/a/main.py", False),
|
||||
],
|
||||
)
|
||||
def test_protected_paths(path: str, protected: bool) -> None:
|
||||
assert is_protected(path) is protected
|
||||
|
||||
|
||||
def test_join() -> None:
|
||||
assert join("", "a.py") == "a.py" and join("s/x", "a.py") == "s/x/a.py"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("text", "expected"),
|
||||
[
|
||||
("```python\nprint(1)\n```", "print(1)"),
|
||||
("```\na\nb\n```", "a\nb"),
|
||||
("print(1)", "print(1)"),
|
||||
("x ``` y", "x ``` y"),
|
||||
],
|
||||
)
|
||||
def test_strip_fence(text: str, expected: str) -> None:
|
||||
assert strip_fence(text) == expected
|
||||
|
||||
|
||||
def test_content_checks() -> None:
|
||||
assert check_content("привет".encode()) == "привет"
|
||||
for data, reason in [(b"a\0b", "binary"), (b"\xff\xfe", "encoding"), (b"x" * (100 * 1024 + 1), "too_big")]:
|
||||
with pytest.raises(ValidationError, match=reason):
|
||||
check_content(data)
|
||||
|
||||
|
||||
def test_python_syntax_only_for_py() -> None:
|
||||
issue = python_syntax("a.py", "def f(:\n pass\n")
|
||||
assert issue is not None and issue.line == 1
|
||||
assert python_syntax("a.py", "x = 1\n") is None
|
||||
assert python_syntax("a.txt", "def f(:") is None
|
||||
|
||||
|
||||
def test_diff_stat() -> None:
|
||||
assert diff_stat("a\nb\nc", "a\nB\nc\nd") == (2, 1)
|
||||
Reference in New Issue
Block a user