91 lines
4.0 KiB
Python
91 lines
4.0 KiB
Python
from datetime import datetime, timezone
|
||
|
||
from aiogram import Bot
|
||
from aiogram.client.session.base import BaseSession
|
||
from aiogram.types import Update, Message, User as TelegramUser, Chat, CallbackQuery
|
||
from sqlalchemy import select
|
||
|
||
from src.core.database import async_session_maker
|
||
from src.core.models import User
|
||
from src.core.services import UserService
|
||
|
||
|
||
class TelegramStub(BaseSession):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.calls = []
|
||
|
||
async def close(self):
|
||
pass
|
||
|
||
async def make_request(self, bot, method, timeout=None):
|
||
self.calls.append(method)
|
||
if method.__api_method__ in {"answerCallbackQuery", "deleteMessage"}:
|
||
return True
|
||
return Message(message_id=100, date=datetime.now(timezone.utc),
|
||
chat=Chat(id=int(getattr(method, "chat_id", 1)), type="private"),
|
||
from_user=TelegramUser(id=bot.id, is_bot=True, first_name="Bot"), text=getattr(method, "text", ""))
|
||
|
||
async def stream_content(self, url, headers=None, timeout=30, chunk_size=65536, raise_for_status=True):
|
||
yield b""
|
||
|
||
|
||
def event(user_id, text=None, callback_data=None):
|
||
actor = TelegramUser(id=user_id, is_bot=False, first_name="Human")
|
||
msg = Message(message_id=1, date=datetime.now(timezone.utc), chat=Chat(id=user_id, type="private"),
|
||
from_user=actor if text else TelegramUser(id=123456, is_bot=True, first_name="Bot"), text=text or "Menu")
|
||
if callback_data:
|
||
return Update(update_id=1, callback_query=CallbackQuery(id="test", from_user=actor, chat_instance="private",
|
||
message=msg, data=callback_data))
|
||
return Update(update_id=1, message=msg)
|
||
|
||
|
||
async def dispatch(update):
|
||
import main
|
||
if not main.dp.sub_routers:
|
||
main.configure_dispatcher()
|
||
stub = TelegramStub()
|
||
bot = Bot("123456:TEST_TOKEN_FOR_ISOLATED_TESTS", session=stub)
|
||
await main.dp.feed_update(bot, update)
|
||
return stub.calls
|
||
|
||
|
||
async def test_actual_back_button_never_registers_bot_as_user():
|
||
calls = await dispatch(event(701, callback_data="back_to_main"))
|
||
async with async_session_maker() as session:
|
||
ids = (await session.scalars(select(User.telegram_id))).all()
|
||
assert ids == [701]
|
||
assert any((getattr(call, "text", "") or "").startswith("👋") for call in calls)
|
||
|
||
|
||
async def test_actual_cashier_menu_available_to_cashier_but_admin_menu_denied():
|
||
calls = await dispatch(event(900002, text="/cashier"))
|
||
assert any("💼 Касса" in (getattr(call, "text", "") or "") for call in calls)
|
||
calls = await dispatch(event(900002, callback_data="admin_participants"))
|
||
assert any("Недостаточно прав" in (getattr(call, "text", "") or "") for call in calls)
|
||
|
||
|
||
async def test_actual_forged_account_callback_is_denied_before_mutation():
|
||
calls = await dispatch(event(702, callback_data="add_to_lottery_1"))
|
||
assert any("Недостаточно прав" in (getattr(call, "text", "") or "") for call in calls)
|
||
|
||
|
||
async def test_assigned_admin_can_open_participant_management():
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_or_create_user(session, 703)
|
||
user.is_admin = True
|
||
await session.commit()
|
||
calls = await dispatch(event(703, callback_data="admin_participants"))
|
||
assert any(getattr(call, "reply_markup", None) for call in calls)
|
||
assert not any("Недостаточно прав" in (getattr(call, "text", "") or "") for call in calls)
|
||
|
||
|
||
async def test_registration_cancel_clears_state():
|
||
calls = await dispatch(event(704, text="/register"))
|
||
assert any("Никнейм" in (getattr(call, "text", "") or "") or "никнейм" in (getattr(call, "text", "") or "") for call in calls)
|
||
import main
|
||
state = main.dp.fsm.get_context(bot=Bot("123456:TEST_TOKEN_FOR_ISOLATED_TESTS"), chat_id=704, user_id=704)
|
||
assert (await state.get_state()).endswith("waiting_for_nickname")
|
||
await dispatch(event(704, text="/cancel"))
|
||
assert await state.get_state() is None
|