# tests/test_handler_guards.py
"""
Guards against stale buttons after an event mode switch, and HTML-escaping of
user/admin-provided text rendered with parse_mode="HTML".
These exercise the actual handler closures registered on the module-level
routers (handlers.user.router / handlers.admin.router), bypassing aiogram's
filter/DI machinery by calling the underlying callback directly — the filters
themselves aren't the concern here, the handler bodies are.
"""
from __future__ import annotations
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import aiosqlite
import pytest
from config import ConferenceEvent, IdleEvent, WebinarEvent, Speaker
from db import init_db
from handlers.user import router as user_router, setup_user_handlers, _send_conference_welcome
from handlers.admin import router as admin_router, setup_admin_handlers
from services.archive import Archive
ADMIN_ID = 999
SPEAKERS = [Speaker(name="Спикер", channel="@chan1")]
def _conference_event(event_id="conf-1", speakers=None):
return ConferenceEvent(event_id=event_id, type="conference", title="T", speakers=speakers or SPEAKERS)
def _webinar_event(event_id="w-1"):
return WebinarEvent(
event_id=event_id,
type="webinar",
title="T",
description="d",
datetime=datetime.fromisoformat("2026-01-01T10:00:00+03:00"),
duration_minutes=30,
)
def _find_handler(router, name, observer="callback_query"):
# setup_user_handlers/setup_admin_handlers register onto the module-level
# router each time they're called (once per `holders` fixture invocation,
# i.e. once per test). Search in reverse so each test gets the handler
# closure bound to *its own* holders, not a stale one from an earlier test.
for h in reversed(getattr(router, observer).handlers):
if h.callback.__name__ == name:
return h.callback
raise LookupError(f"handler {name!r} not registered on router.{observer}")
class FakeMessage:
def __init__(self, document=None):
self.answer = AsyncMock()
self.document = document
class FakeCallback:
def __init__(self, data, user_id=1):
self.data = data
self.from_user = SimpleNamespace(id=user_id)
self.message = FakeMessage()
self.answer = AsyncMock()
@pytest.fixture
async def holders():
"""Register a fresh set of handler closures (own event_holder + real
in-memory db) for each test. See _find_handler for why this must be
per-test rather than shared."""
async with aiosqlite.connect(":memory:") as db:
await init_db(db)
event_holder = {"event": _conference_event()}
welcome_holder = {"welcome": None}
archive_holder = {"archive": Archive(), "path": "archive.json"}
setup_user_handlers(db=db, event_holder=event_holder, welcome_holder=welcome_holder, archive_holder=archive_holder)
setup_admin_handlers(
db=db,
event_holder=event_holder,
event_path="event.json",
admin_id=ADMIN_ID,
welcome_holder=welcome_holder,
welcome_path="welcome.json",
archive_holder=archive_holder,
)
yield {"event_holder": event_holder, "db": db}
# --- Guard: check_subscriptions callback after event mode switch ---
async def test_check_subscriptions_guards_against_non_conference_event(holders):
holders["event_holder"]["event"] = IdleEvent(event_id="idle-1", type="idle", title="Архив")
handler = _find_handler(user_router, "on_check_subscriptions")
bot = AsyncMock()
cb = FakeCallback(data="check_subscriptions")
await handler(cb, bot=bot)
bot.get_chat_member.assert_not_called()
cb.message.answer.assert_called_once()
text = cb.message.answer.call_args.args[0]
assert "Событие сменилось" in text
assert "/start" in text
async def test_check_subscriptions_still_works_for_conference_event(holders):
from db import record_start
holders["event_holder"]["event"] = _conference_event()
await record_start(holders["db"], user_id=1, conference_id="conf-1")
handler = _find_handler(user_router, "on_check_subscriptions")
bot = AsyncMock()
bot.get_chat_member = AsyncMock(return_value=SimpleNamespace(status="member"))
cb = FakeCallback(data="check_subscriptions")
await handler(cb, bot=bot)
bot.get_chat_member.assert_called()
text = cb.message.answer.call_args.args[0]
assert "подписан на всех спикеров" in text
async def test_check_subscriptions_escapes_missing_speaker_name(holders):
evil = Speaker(name="Evil", channel="@evil")
holders["event_holder"]["event"] = _conference_event(speakers=[evil])
handler = _find_handler(user_router, "on_check_subscriptions")
bot = AsyncMock()
bot.get_chat_member = AsyncMock(return_value=SimpleNamespace(status="left"))
cb = FakeCallback(data="check_subscriptions")
await handler(cb, bot=bot)
text = cb.message.answer.call_args.args[0]
assert "Evil" not in text
assert "<b>Evil</b>" in text
# --- Guard: broadcast confirmation after event mode switch (admin.py) ---
async def test_broadcast_confirm_guards_against_non_conference_event(holders):
holders["event_holder"]["event"] = _webinar_event()
handler = _find_handler(admin_router, "on_broadcast_confirm")
state = MagicMock()
state.get_data = AsyncMock(return_value={"text": "Hi", "photo_file_id": None})
state.clear = AsyncMock()
bot = AsyncMock()
cb = FakeCallback(data="broadcast_confirm", user_id=ADMIN_ID)
with patch("handlers.admin.broadcast", new=AsyncMock()) as mock_broadcast:
await handler(cb, state=state, bot=bot)
mock_broadcast.assert_not_called()
texts = [c.args[0] for c in cb.message.answer.call_args_list]
assert any("Событие изменилось, рассылка отменена" in t for t in texts)
assert not any("Рассылка запущена" in t for t in texts)
async def test_broadcast_confirm_proceeds_for_conference_event(holders):
holders["event_holder"]["event"] = _conference_event()
handler = _find_handler(admin_router, "on_broadcast_confirm")
state = MagicMock()
state.get_data = AsyncMock(return_value={"text": "Hi", "photo_file_id": None})
state.clear = AsyncMock()
bot = AsyncMock()
cb = FakeCallback(data="broadcast_confirm", user_id=ADMIN_ID)
with patch("handlers.admin.broadcast", new=AsyncMock(return_value={"delivered": 0, "skipped": 0, "errors": 0})) as mock_broadcast:
await handler(cb, state=state, bot=bot)
mock_broadcast.assert_called_once()
# --- Guard: event.json download failures don't crash the handler (admin.py) ---
async def test_event_json_download_failure_replies_instead_of_raising(holders):
handler = _find_handler(admin_router, "on_event_json", observer="message")
bot = AsyncMock()
bot.get_file = AsyncMock(side_effect=Exception("network down"))
message = FakeMessage(document=SimpleNamespace(file_name="event.json", file_id="abc"))
await handler(message, bot=bot) # must not raise
message.answer.assert_called_once()
assert "Не удалось загрузить файл" in message.answer.call_args.args[0]
# --- HTML escaping ---
async def test_send_conference_welcome_escapes_speaker_name():
event = _conference_event(speakers=[Speaker(name="", channel="@x")])
message = FakeMessage()
await _send_conference_welcome(message, event, welcome=None)
text = message.answer.call_args.args[0]
assert "" not in text
assert "<script>alert(1)</script>" in text