Files
ElebobaandClaude Opus 5.5 6c6f766abc
CI / ruff · mypy · pytest (pull_request) Successful in 4m49s
feat(alerts): алерты всей команде — детекторы, дедупликация, список (#15)
- детекторы по состоянию после каждого раунда снапшотов: краш, OOM,
  цикл рестартов, unhealthy, перегрузка ≥ 1 мин, мало места, Docker
  недоступен; пороги в config/alerts.yaml
- AlertService: ключ kind+service, повтор в 10 мин — счётчик в исходном
  сообщении, «✅ восстановлено» ответом; разовые события (🛡) со счётчиком
- рассылка всем активным (🛡 — только админам), темп ~25/с, RetryAfter,
  заблокировавший бота пропускается; alert_deliveries для правок и ответов
- «🚨 Алерты»: активные + 7 дней по 10, карточка; кнопки «📜 Логи»,
  «🩺 Статус» под алертом открывают новое сообщение, а не затирают алерт
- 🛡 атаки через агента идут через общий сервис алертов
- убраны заглушки «Раздел в разработке» — у всех разделов свои роутеры

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 15:36:09 +03:00

67 lines
2.4 KiB
Python

import pytest
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage
from aiogram.types import ReplyKeyboardMarkup
from app.access.service import AccessService
from app.bot.menu import Nav, NavAction
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, OUTSIDER
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 buttons(markup: ReplyKeyboardMarkup) -> list[str]:
return [b.text for row in markup.keyboard for b in row]
async def test_admin_start_shows_welcome_and_full_menu(tg: TgClient, texts: Texts) -> None:
await tg.send(ADMIN, "/start")
[reply] = tg.session.sent(SendMessage)
assert reply.text == texts.start.welcome
assert isinstance(reply.reply_markup, ReplyKeyboardMarkup)
m = texts.menu
assert buttons(reply.reply_markup) == [m.commit, m.status, m.logs, m.alerts, m.members]
async def test_outsider_gets_one_refusal_then_silence(tg: TgClient, texts: Texts) -> None:
await tg.send(OUTSIDER, "/start")
await tg.send(OUTSIDER, "hello")
await tg.send(OUTSIDER, texts.menu.commit)
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.start.no_access]
async def test_outsider_stale_button_is_swallowed(tg: TgClient) -> None:
await tg.press(OUTSIDER, Nav(action=NavAction.BACK).pack())
assert tg.session.sent(EditMessageText) == []
assert len(tg.session.sent(AnswerCallbackQuery)) == 1
@pytest.mark.parametrize(
("action", "expected"),
[(NavAction.BACK, "placeholder"), (NavAction.CANCEL, "cancelled")],
)
async def test_nav_edits_section_message_in_place(tg: TgClient, texts: Texts, action: NavAction, expected: str) -> None:
await tg.press(ADMIN, Nav(action=action).pack())
[edit] = tg.session.sent(EditMessageText)
want = texts.menu.placeholder if expected == "placeholder" else texts.section.cancelled
assert edit.text == want
assert tg.session.sent(SendMessage) == []
async def test_menu_and_help_commands(tg: TgClient, texts: Texts) -> None:
await tg.send(ADMIN, "/menu")
await tg.send(ADMIN, "/help")
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.menu.placeholder, texts.help.text]