From 8d2222fd611e9d90d1ca709d6ad8c3ae3c190530 Mon Sep 17 00:00:00 2001 From: Mohammad Date: Tue, 12 May 2026 21:30:35 +0330 Subject: [PATCH 1/7] test: add comprehensive test suite for Telegram handlers --- tests/telegram/conftest.py | 122 ++++++ .../handlers/admin/test_bulk_actions.py | 226 ++++++++++ .../handlers/admin/test_confirm_action.py | 60 +++ .../telegram/handlers/admin/test_main_menu.py | 92 +++++ tests/telegram/handlers/admin/test_user.py | 391 ++++++++++++++++++ .../handlers/client/test_show_info.py | 60 +++ tests/telegram/handlers/test_base.py | 71 ++++ tests/telegram/handlers/test_error_handler.py | 37 ++ 8 files changed, 1059 insertions(+) create mode 100644 tests/telegram/conftest.py create mode 100644 tests/telegram/handlers/admin/test_bulk_actions.py create mode 100644 tests/telegram/handlers/admin/test_confirm_action.py create mode 100644 tests/telegram/handlers/admin/test_main_menu.py create mode 100644 tests/telegram/handlers/admin/test_user.py create mode 100644 tests/telegram/handlers/client/test_show_info.py create mode 100644 tests/telegram/handlers/test_base.py create mode 100644 tests/telegram/handlers/test_error_handler.py diff --git a/tests/telegram/conftest.py b/tests/telegram/conftest.py new file mode 100644 index 000000000..76d6ecf69 --- /dev/null +++ b/tests/telegram/conftest.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from app.db.models import UserStatus +from app.models.admin import AdminDetails + + +@dataclass +class FakeChat: + id: int = 1 + + +@dataclass +class FakeFromUser: + full_name: str = "Test User" + + +class FakeState: + def __init__(self): + self._data: dict = {} + self._state = None + + async def set_state(self, state): + self._state = state + + async def get_state(self): + return self._state + + async def clear(self): + self._state = None + self._data = {} + + async def update_data(self, **kwargs): + self._data.update(kwargs) + + async def get_data(self): + return dict(self._data) + + async def get_value(self, key, default=None): + return self._data.get(key, default) + + +class FakeMessage: + def __init__(self, text: str = "text", message_id: int = 100, chat_id: int = 1): + self.text = text + self.message_id = message_id + self.chat = FakeChat(chat_id) + self.from_user = FakeFromUser() + self.bot = SimpleNamespace(delete_messages=AsyncMock()) + self.answer = AsyncMock(side_effect=self._new_message) + self.reply = AsyncMock(side_effect=self._new_message) + self.edit_text = AsyncMock() + self.edit_reply_markup = AsyncMock() + self.answer_document = AsyncMock() + self.delete = AsyncMock() + + async def _new_message(self, *args, **kwargs): + return FakeMessage(message_id=self.message_id + 1, chat_id=self.chat.id) + + +class FakeCallbackQuery: + def __init__(self, message: FakeMessage | None = None, data: str = "callback"): + self.message = message or FakeMessage() + self.data = data + self.from_user = self.message.from_user + self.bot = self.message.bot + self.answer = AsyncMock() + + +class FakeInlineQuery: + def __init__(self, query: str = ""): + self.query = query + self.answer = AsyncMock() + + +@pytest.fixture +def fake_state() -> FakeState: + return FakeState() + + +@pytest.fixture +def fake_message() -> FakeMessage: + return FakeMessage() + + +@pytest.fixture +def fake_callback() -> FakeCallbackQuery: + return FakeCallbackQuery() + + +@pytest.fixture +def fake_inline_query() -> FakeInlineQuery: + return FakeInlineQuery() + + +@pytest.fixture +def admin() -> AdminDetails: + return AdminDetails(username="testadmin", is_sudo=True) + + +@pytest.fixture +def regular_admin() -> AdminDetails: + return AdminDetails(username="owner", is_sudo=False) + + +@pytest.fixture +def fake_user(): + user = SimpleNamespace( + id=11, + username="alice", + status=UserStatus.active, + subscription_url="https://example.com/sub/alice", + next_plan=None, + admin=SimpleNamespace(username="owner"), + ) + user.model_dump = lambda: {} + return user diff --git a/tests/telegram/handlers/admin/test_bulk_actions.py b/tests/telegram/handlers/admin/test_bulk_actions.py new file mode 100644 index 000000000..a051418d5 --- /dev/null +++ b/tests/telegram/handlers/admin/test_bulk_actions.py @@ -0,0 +1,226 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from app.models.user import UsernameGenerationStrategy +from app.telegram.handlers.admin import bulk_actions as bulk_h +from app.telegram.keyboards.bulk_actions import ( + BulkAction, + BulkActionPanel, + BulkTemplateSelector, + UsernameStrategySelector, +) + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_event_types(monkeypatch, fake_message, fake_callback): + monkeypatch.setattr(bulk_h, "Texts", _DummyTexts()) + monkeypatch.setattr(bulk_h, "Message", type(fake_message)) + monkeypatch.setattr(bulk_h, "CallbackQuery", type(fake_callback)) + monkeypatch.setattr(bulk_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(bulk_h, "add_to_messages_to_delete", AsyncMock()) + + +@pytest.mark.asyncio +async def test_helpers_message_target_and_chunking(fake_message, fake_callback): + msg = type(fake_message)() + callback = type(fake_callback)(message=msg) + + assert bulk_h._message_target(msg) is msg + assert bulk_h._message_target(callback) is msg + chunks = bulk_h._chunk_subscription_urls(["a" * 3000, "b" * 3000], limit=3800) + assert len(chunks) == 2 + + +@pytest.mark.asyncio +async def test_bulk_actions_menu(fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await bulk_h.bulk_actions(event) + + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bulk_create_from_template_no_templates(monkeypatch, fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + monkeypatch.setattr(bulk_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) + + await bulk_h.bulk_create_from_template(event, db=object(), state=fake_state) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bulk_template_chosen(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = BulkTemplateSelector.Callback(template_id=1) + + await bulk_h.bulk_template_chosen(event, state=fake_state, callback_data=callback_data) + + assert await fake_state.get_state() == bulk_h.forms.BulkCreateFromTemplate.count + + +@pytest.mark.asyncio +async def test_bulk_template_count_invalid(fake_state, fake_message): + event = type(fake_message)(text="x") + + await bulk_h.bulk_template_count(event, state=fake_state) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bulk_template_strategy_random(monkeypatch, fake_state, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UsernameStrategySelector.Callback(strategy=UsernameGenerationStrategy.random) + fake_state._data = {"template_id": 1, "count": 2} + monkeypatch.setattr(bulk_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(bulk_h, "_perform_bulk_creation", AsyncMock()) + + await bulk_h.bulk_template_strategy(event, db=object(), state=fake_state, admin=admin, callback_data=callback_data) + + bulk_h._perform_bulk_creation.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bulk_template_sequence_username_invalid(fake_state, admin, fake_message): + event = type(fake_message)(text="bad name") + + await bulk_h.bulk_template_sequence_username(event, db=object(), state=fake_state, admin=admin) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bulk_template_start_number_invalid(fake_state, admin, fake_message): + event = type(fake_message)(text="-2") + + await bulk_h.bulk_template_start_number(event, db=object(), state=fake_state, admin=admin) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_perform_bulk_creation_success(monkeypatch, fake_state, admin, fake_message): + event = type(fake_message)(text="run") + monkeypatch.setattr(bulk_h, "delete_messages", AsyncMock()) + monkeypatch.setattr( + bulk_h.user_operations, + "bulk_create_users_from_template", + AsyncMock(return_value=SimpleNamespace(created=2, subscription_urls=["https://a", "https://b"])), + ) + + await bulk_h._perform_bulk_creation( + event, + db=object(), + admin=admin, + state=fake_state, + template_id=1, + count=2, + strategy=UsernameGenerationStrategy.random, + ) + + assert event.answer.await_count >= 2 + + +@pytest.mark.asyncio +async def test_delete_expired_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await bulk_h.delete_expired(event, state=fake_state) + + assert await fake_state.get_state() == bulk_h.forms.DeleteExpired.expired_before + + +@pytest.mark.asyncio +async def test_process_expire_before_validation(fake_state, fake_message): + event = type(fake_message)(text="abc") + + await bulk_h.process_expire_before(event, state=fake_state) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delete_expired_done(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = BulkActionPanel.Callback(action=BulkAction.delete_expired, amount="3") + monkeypatch.setattr( + bulk_h.user_operations, + "delete_expired_users", + AsyncMock(return_value=SimpleNamespace(count=5)), + ) + + await bulk_h.delete_expired_done(event, db=object(), admin=admin, callback_data=callback_data) + + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_expiry_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await bulk_h.modify_expiry(event, state=fake_state) + + assert await fake_state.get_state() == bulk_h.forms.BulkModify.expiry + + +@pytest.mark.asyncio +async def test_process_expiry_invalid(fake_state, fake_message): + event = type(fake_message)(text="bad") + + await bulk_h.process_expiry(event, state=fake_state) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_expiry_done(monkeypatch, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = BulkActionPanel.Callback(action=BulkAction.modify_expiry, amount="5") + monkeypatch.setattr(bulk_h.user_operations, "bulk_modify_expire", AsyncMock(return_value=7)) + + await bulk_h.modify_expiry_done(event, db=object(), callback_data=callback_data) + + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_data_limit_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await bulk_h.modify_data_limit(event, state=fake_state) + + assert await fake_state.get_state() == bulk_h.forms.BulkModify.data_limit + + +@pytest.mark.asyncio +async def test_process_data_limit_invalid(fake_state, fake_message): + event = type(fake_message)(text="bad") + + await bulk_h.process_data_limit(event, state=fake_state) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_data_limit_done(monkeypatch, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = BulkActionPanel.Callback(action=BulkAction.modify_data_limit, amount="4") + monkeypatch.setattr(bulk_h.user_operations, "bulk_modify_datalimit", AsyncMock(return_value=9)) + + await bulk_h.modify_data_limit_done(event, db=object(), callback_data=callback_data) + + event.message.edit_text.assert_awaited_once() diff --git a/tests/telegram/handlers/admin/test_confirm_action.py b/tests/telegram/handlers/admin/test_confirm_action.py new file mode 100644 index 000000000..8bad561ad --- /dev/null +++ b/tests/telegram/handlers/admin/test_confirm_action.py @@ -0,0 +1,60 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from app.telegram.handlers.admin import confirm_action +from app.telegram.keyboards.user import UserPanel, UserPanelAction + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_texts(monkeypatch): + monkeypatch.setattr(confirm_action, "Texts", _DummyTexts()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "action", + [ + UserPanelAction.disable, + UserPanelAction.enable, + UserPanelAction.delete, + UserPanelAction.revoke_sub, + UserPanelAction.reset_usage, + UserPanelAction.activate_next_plan, + ], +) +async def test_confirm_action_for_user_actions(monkeypatch, admin, action, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + user = SimpleNamespace(username="alice") + callback = SimpleNamespace( + action=UserPanel.Callback(user_id=11, action=action).pack(), + cancel=UserPanel.Callback(user_id=11).pack(), + ) + + monkeypatch.setattr(confirm_action.user_operations, "get_user_by_id", AsyncMock(return_value=user)) + + await confirm_action.confirm_action(event, callback, db=object(), admin=admin) + + confirm_action.user_operations.get_user_by_id.assert_awaited_once() + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_confirm_action_default_text_without_user_action(admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback = SimpleNamespace(action="other:payload", cancel="cancel") + + await confirm_action.confirm_action(event, callback, db=object(), admin=admin) + + event.message.edit_text.assert_awaited_once() diff --git a/tests/telegram/handlers/admin/test_main_menu.py b/tests/telegram/handlers/admin/test_main_menu.py new file mode 100644 index 000000000..7ee8e7cbf --- /dev/null +++ b/tests/telegram/handlers/admin/test_main_menu.py @@ -0,0 +1,92 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from aiogram.exceptions import TelegramBadRequest + +from app.telegram.handlers.admin import main_menu + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_texts(monkeypatch): + monkeypatch.setattr(main_menu, "Texts", _DummyTexts()) + + +@pytest.mark.asyncio +async def test_reload_data(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + monkeypatch.setattr(main_menu.system_operator, "get_system_stats", AsyncMock(return_value={})) + monkeypatch.setattr( + main_menu, + "telegram_settings", + AsyncMock(return_value=SimpleNamespace(mini_app_login=True, mini_app_web_url="https://panel")), + ) + + await main_menu.reload_data(event, db=object(), admin=admin) + + event.message.edit_text.assert_awaited_once() + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_reload_data_ignores_edit_error(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + event.message.edit_text = AsyncMock(side_effect=TelegramBadRequest(method="m", message="bad")) + monkeypatch.setattr(main_menu.system_operator, "get_system_stats", AsyncMock(return_value={})) + monkeypatch.setattr( + main_menu, + "telegram_settings", + AsyncMock(return_value=SimpleNamespace(mini_app_login=False, mini_app_web_url="https://panel")), + ) + + await main_menu.reload_data(event, db=object(), admin=admin) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_sync_users(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + nodes = SimpleNamespace(nodes=[SimpleNamespace(id=1), SimpleNamespace(id=2)]) + + monkeypatch.setattr(main_menu.node_operator, "get_db_nodes", AsyncMock(return_value=nodes)) + monkeypatch.setattr(main_menu.node_operator, "sync_node_users", AsyncMock()) + monkeypatch.setattr(main_menu.system_operator, "get_system_stats", AsyncMock(return_value={})) + monkeypatch.setattr( + main_menu, + "telegram_settings", + AsyncMock(return_value=SimpleNamespace(mini_app_login=False, mini_app_web_url="https://panel")), + ) + + await main_menu.sync_users(event, db=object(), admin=admin) + + assert main_menu.node_operator.sync_node_users.await_count == 2 + assert event.answer.await_count == 2 + + +@pytest.mark.asyncio +async def test_reconnect_all_nodes(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + monkeypatch.setattr(main_menu.node_operator, "restart_all_node", AsyncMock()) + monkeypatch.setattr(main_menu.system_operator, "get_system_stats", AsyncMock(return_value={})) + monkeypatch.setattr( + main_menu, + "telegram_settings", + AsyncMock(return_value=SimpleNamespace(mini_app_login=False, mini_app_web_url="https://panel")), + ) + + await main_menu.reconnect_all_nodes(event, db=object(), admin=admin) + + main_menu.node_operator.restart_all_node.assert_awaited_once() + assert event.answer.await_count == 2 diff --git a/tests/telegram/handlers/admin/test_user.py b/tests/telegram/handlers/admin/test_user.py new file mode 100644 index 000000000..e6d0ee52e --- /dev/null +++ b/tests/telegram/handlers/admin/test_user.py @@ -0,0 +1,391 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from app.db.models import UserStatus +from app.telegram.handlers.admin import user as user_h +from app.telegram.keyboards.group import GroupsSelector, SelectGroupAction +from app.telegram.keyboards.user import ChooseStatus, ChooseTemplate, UserPanel, UserPanelAction + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_event_types(monkeypatch, fake_message, fake_callback, fake_inline_query): + monkeypatch.setattr(user_h, "Texts", _DummyTexts()) + monkeypatch.setattr(user_h, "Message", type(fake_message)) + monkeypatch.setattr(user_h, "CallbackQuery", type(fake_callback)) + monkeypatch.setattr(user_h, "InlineQuery", type(fake_inline_query)) + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + + +@pytest.fixture +def groups_response(): + return SimpleNamespace(groups=[SimpleNamespace(id=1, name="g1"), SimpleNamespace(id=2, name="g2")]) + + +@pytest.mark.asyncio +async def test_create_user(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await user_h.create_user(event, fake_state) + + assert await fake_state.get_state() == user_h.forms.CreateUser.username + event.message.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_username_message_happy(monkeypatch, fake_state, admin, fake_message): + event = type(fake_message)(text="alice") + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + monkeypatch.setattr(user_h.UserValidator, "validate_username", lambda x: x) + monkeypatch.setattr(user_h.user_operations, "get_validated_user", AsyncMock(side_effect=ValueError("not found"))) + + await user_h.process_username(event, fake_state, db=object(), admin=admin) + + assert await fake_state.get_state() == user_h.forms.CreateUser.data_limit + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_username_callback_duplicate(monkeypatch, fake_state, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + monkeypatch.setattr(user_h.random, "choices", lambda *args, **kwargs: list("ABCDE")) + monkeypatch.setattr(user_h.user_operations, "get_validated_user", AsyncMock(return_value=SimpleNamespace())) + + await user_h.process_username(event, fake_state, db=object(), admin=admin) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_data_limit_validation(fake_state, fake_message): + event = type(fake_message)(text="-1") + + await user_h.process_data_limit(event, fake_state) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_expire_zero(monkeypatch, fake_state, groups_response, fake_message): + event = type(fake_message)(text="0") + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + monkeypatch.setattr(user_h.group_operations, "get_all_groups", AsyncMock(return_value=groups_response)) + + await user_h.process_expire(event, fake_state, db=object()) + + assert await fake_state.get_state() == user_h.forms.CreateUser.group_ids + + +@pytest.mark.asyncio +async def test_process_status_on_hold(monkeypatch, fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = ChooseStatus.Callback(status=UserStatus.on_hold.value) + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + + await user_h.process_status(event, db=object(), state=fake_state, callback_data=callback_data) + + assert await fake_state.get_state() == user_h.forms.CreateUser.on_hold_timeout + + +@pytest.mark.asyncio +async def test_process_on_hold_timeout_invalid(fake_state, fake_message): + event = type(fake_message)(text="bad") + + await user_h.process_on_hold_timeout(event, fake_state, db=object()) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_select_groups_toggles(monkeypatch, fake_state, groups_response, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = GroupsSelector.Callback(action=SelectGroupAction.select, group_id=1, user_id=0) + monkeypatch.setattr(user_h.group_operations, "get_all_groups", AsyncMock(return_value=groups_response)) + + await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data) + await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data) + + assert await fake_state.get_value("group_ids") == [] + + +@pytest.mark.asyncio +async def test_process_done_without_groups(fake_state, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await user_h.process_done(event, db=object(), admin=admin, state=fake_state) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_done_creates_user(monkeypatch, fake_state, admin, fake_user, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + fake_state._data = { + "username": "alice", + "data_limit": 1.5, + "duration": 3, + "status": UserStatus.active.value, + "group_ids": [1], + "messages_to_delete": [1], + } + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h.user_operations, "create_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_all_groups", AsyncMock(return_value=[])) + + await user_h.process_done(event, db=object(), admin=admin, state=fake_state) + + user_h.user_operations.create_user.assert_awaited_once() + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_groups_user_not_found(monkeypatch, fake_state, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=1, action=UserPanelAction.modify_groups) + monkeypatch.setattr(user_h.user_operations, "get_user_by_id", AsyncMock(side_effect=ValueError("missing"))) + + await user_h.modify_groups(event, db=object(), state=fake_state, callback_data=callback_data, admin=admin) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_groups_done_no_groups(fake_state, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + + await user_h.modify_groups_done(event, db=object(), admin=admin, state=fake_state) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_expiry_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=1, action=UserPanelAction.modify_expiry) + + await user_h.modify_expiry(event, callback_data=callback_data, state=fake_state) + + assert await fake_state.get_state() == user_h.forms.ModifyUser.new_expiry + + +@pytest.mark.asyncio +async def test_modify_expiry_done_invalid(fake_state, admin, fake_message): + event = type(fake_message)(text="bad") + + await user_h.modify_expiry_done(event, state=fake_state, db=object(), admin=admin) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_data_limit_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=1, action=UserPanelAction.modify_data_limit) + + await user_h.modify_data_limit(event, callback_data=callback_data, state=fake_state) + + assert await fake_state.get_state() == user_h.forms.ModifyUser.new_data_limit + + +@pytest.mark.asyncio +async def test_modify_data_limit_done_invalid(fake_state, admin, fake_message): + event = type(fake_message)(text="abc") + + await user_h.modify_data_limit_done(event, state=fake_state, db=object(), admin=admin) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_note_sets_state(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=1, action=UserPanelAction.modify_note) + + await user_h.modify_note(event, callback_data=callback_data, state=fake_state) + + assert await fake_state.get_state() == user_h.forms.ModifyUser.new_note + + +@pytest.mark.asyncio +async def test_modify_note_done_user_not_found(monkeypatch, fake_state, admin, fake_message): + event = type(fake_message)(text="note") + fake_state._data = {"user_id": 10} + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + monkeypatch.setattr(user_h.user_operations, "get_user_by_id", AsyncMock(side_effect=ValueError("missing"))) + + await user_h.modify_note_done(event, state=fake_state, db=object(), admin=admin) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "func_name,action", + [ + ("disable_user", UserPanelAction.disable), + ("enable_user", UserPanelAction.enable), + ("delete_user", UserPanelAction.delete), + ("revoke_sub", UserPanelAction.revoke_sub), + ("reset_usage", UserPanelAction.reset_usage), + ("activate_next_plan", UserPanelAction.activate_next_plan), + ], +) +async def test_direct_user_actions(monkeypatch, admin, fake_user, func_name, action, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=action) + + monkeypatch.setattr(user_h.user_operations, "get_user_by_id", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_all_groups", AsyncMock(return_value=[])) + monkeypatch.setattr(user_h.user_operations, "modify_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "remove_user", AsyncMock()) + monkeypatch.setattr(user_h.user_operations, "revoke_user_sub", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "reset_user_data_usage", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "active_next_plan", AsyncMock(return_value=fake_user)) + + await getattr(user_h, func_name)(event, admin=admin, db=object(), callback_data=callback_data) + + assert event.answer.await_count >= 1 + + +@pytest.mark.asyncio +async def test_modify_with_template_no_templates(monkeypatch, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.modify_with_template) + monkeypatch.setattr(user_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) + + await user_h.modify_with_template(event, db=object(), callback_data=callback_data) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_modify_with_template_done(monkeypatch, admin, fake_user, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = ChooseTemplate.Callback(template_id=1, user_id=11) + monkeypatch.setattr(user_h.user_operations, "get_user_by_id", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "modify_user_with_template", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_all_groups", AsyncMock(return_value=[])) + + await user_h.modify_with_template_done(event, db=object(), admin=admin, callback_data=callback_data) + + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_create_user_from_template_no_templates(monkeypatch, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + monkeypatch.setattr(user_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) + + await user_h.create_user_from_template(event, db=object()) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_create_user_from_template_username(fake_state, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = ChooseTemplate.Callback(template_id=1) + + await user_h.create_user_from_template_username(event, state=fake_state, callback_data=callback_data) + + assert await fake_state.get_state() == user_h.forms.CreateUserFromTemplate.username + + +@pytest.mark.asyncio +async def test_create_user_from_template_choose(monkeypatch, fake_state, admin, fake_user, fake_message): + event = type(fake_message)(text="alice") + fake_state._data = {"template_id": 1, "messages_to_delete": []} + template = SimpleNamespace(username_prefix="", username_suffix="") + + monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) + monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) + monkeypatch.setattr(user_h.UserValidator, "validate_username", lambda x: x) + monkeypatch.setattr(user_h.user_templates, "get_validated_user_template", AsyncMock(return_value=template)) + monkeypatch.setattr(user_h.user_operations, "get_validated_user", AsyncMock(side_effect=ValueError("missing"))) + monkeypatch.setattr(user_h.user_operations, "create_user_from_template", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_all_groups", AsyncMock(return_value=[])) + + await user_h.create_user_from_template_choose(event, state=fake_state, db=object(), admin=admin) + + user_h.user_operations.create_user_from_template.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_user_by_sub_unauthorized(monkeypatch, regular_admin, fake_user, fake_message): + event = type(fake_message)(text="/sub/abc") + fake_user.admin = SimpleNamespace(username="other") + + monkeypatch.setattr(user_h.user_operations, "get_validated_sub", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_user", AsyncMock(return_value=fake_user)) + + await user_h.get_user_by_sub(event, db=object(), admin=regular_admin) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_v2ray_links_unavailable(monkeypatch, admin, fake_user, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.v2ray_links) + monkeypatch.setattr(user_h.user_operations, "get_validated_user_by_id", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.subscription_operations, "validated_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.subscription_operations, "fetch_config", AsyncMock(return_value=("", None))) + + await user_h.get_v2ray_links(event, db=object(), admin=admin, callback_data=callback_data) + + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_user_message_not_found(monkeypatch, admin, fake_message): + event = type(fake_message)(text="missing") + monkeypatch.setattr(user_h.user_operations, "get_user", AsyncMock(side_effect=ValueError("missing"))) + + await user_h.get_user(event, admin=admin, db=object()) + + event.reply.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_user_callback_happy(monkeypatch, admin, fake_user, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.show) + monkeypatch.setattr(user_h.user_operations, "get_user_by_id", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(user_h.user_operations, "validate_all_groups", AsyncMock(return_value=[])) + + await user_h.get_user(event, admin=admin, db=object(), callback_data=callback_data) + + event.message.edit_text.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_search_user_and_debug(monkeypatch, admin, fake_user, fake_inline_query, fake_message, fake_callback): + inline = type(fake_inline_query)(query="al") + callback = type(fake_callback)(message=type(fake_message)(), data="any:payload") + + monkeypatch.setattr(user_h.user_operations, "get_users", AsyncMock(return_value=SimpleNamespace(users=[fake_user]))) + + await user_h.search_user(inline, admin=admin, db=object()) + await user_h.debug(callback) + + inline.answer.assert_awaited_once() + callback.answer.assert_awaited_once_with("any:payload", show_alert=True) diff --git a/tests/telegram/handlers/client/test_show_info.py b/tests/telegram/handlers/client/test_show_info.py new file mode 100644 index 000000000..ff589181b --- /dev/null +++ b/tests/telegram/handlers/client/test_show_info.py @@ -0,0 +1,60 @@ +from unittest.mock import AsyncMock + +import pytest + +from app.telegram.handlers.client import show_info + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_texts(monkeypatch): + monkeypatch.setattr(show_info, "Texts", _DummyTexts()) + + +@pytest.mark.asyncio +async def test_get_user_client_links_as_text(monkeypatch, fake_user, fake_message): + event = type(fake_message)(text="/sub/token") + + monkeypatch.setattr(show_info.user_operations, "get_validated_sub", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.user_operations, "validate_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.subscription_operations, "validated_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.subscription_operations, "fetch_config", AsyncMock(return_value=("vmess://a", None))) + + await show_info.get_user(event, db=object()) + + assert event.reply.await_count == 1 + assert event.answer.await_count == 1 + + +@pytest.mark.asyncio +async def test_get_user_client_links_as_file(monkeypatch, fake_user, fake_message): + event = type(fake_message)(text="/sub/token") + + monkeypatch.setattr(show_info.user_operations, "get_validated_sub", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.user_operations, "validate_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.subscription_operations, "validated_user", AsyncMock(return_value=fake_user)) + monkeypatch.setattr(show_info.subscription_operations, "fetch_config", AsyncMock(return_value=("x" * 5000, None))) + + await show_info.get_user(event, db=object()) + + event.answer_document.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_user_client_not_found(monkeypatch, fake_message): + event = type(fake_message)(text="/sub/token") + + monkeypatch.setattr(show_info.user_operations, "get_validated_sub", AsyncMock(side_effect=ValueError("missing"))) + + await show_info.get_user(event, db=object()) + + event.reply.assert_awaited_once() diff --git a/tests/telegram/handlers/test_base.py b/tests/telegram/handlers/test_base.py new file mode 100644 index 000000000..5c02606f4 --- /dev/null +++ b/tests/telegram/handlers/test_base.py @@ -0,0 +1,71 @@ +from unittest.mock import AsyncMock + +import pytest +from aiogram.exceptions import TelegramBadRequest + +from app.telegram.handlers import base + + +class _TextValue(str): + def __call__(self, *args, **kwargs): + return str(self) + + +class _DummyTexts: + def __getattr__(self, name): + return _TextValue(name) + + +@pytest.fixture(autouse=True) +def patch_texts(monkeypatch, fake_message, fake_callback): + monkeypatch.setattr(base, "Texts", _DummyTexts()) + monkeypatch.setattr(base.types, "Message", type(fake_message)) + monkeypatch.setattr(base.types, "CallbackQuery", type(fake_callback)) + + +@pytest.mark.asyncio +async def test_command_start_handler_for_admin_message(monkeypatch, admin, fake_state, fake_message): + event = type(fake_message)(text="/start") + fake_state._state = "busy" + fake_state._data = {"messages_to_delete": [10]} + + monkeypatch.setattr(base, "delete_messages", AsyncMock()) + monkeypatch.setattr(base.system_operator, "get_system_stats", AsyncMock(return_value={"ok": True})) + monkeypatch.setattr( + base, + "telegram_settings", + AsyncMock(return_value=type("S", (), {"mini_app_login": True, "mini_app_web_url": "https://panel"})()), + ) + + await base.command_start_handler(event, admin, fake_state, db=object()) + + base.delete_messages.assert_awaited_once() + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_command_start_handler_callback_falls_back_to_answer(monkeypatch, admin, fake_message, fake_callback): + message = type(fake_message)() + message.edit_text = AsyncMock(side_effect=TelegramBadRequest(method="m", message="bad")) + event = type(fake_callback)(message=message) + + monkeypatch.setattr(base.system_operator, "get_system_stats", AsyncMock(return_value={})) + monkeypatch.setattr( + base, + "telegram_settings", + AsyncMock(return_value=type("S", (), {"mini_app_login": False, "mini_app_web_url": "https://panel"})()), + ) + + await base.command_start_handler(event, admin, None, db=object()) + + event.message.edit_text.assert_awaited_once() + event.message.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_command_start_handler_non_admin(fake_message): + event = type(fake_message)(text="/start") + + await base.command_start_handler(event, None, None, db=None) + + event.answer.assert_awaited_once() diff --git a/tests/telegram/handlers/test_error_handler.py b/tests/telegram/handlers/test_error_handler.py new file mode 100644 index 000000000..90493a5ef --- /dev/null +++ b/tests/telegram/handlers/test_error_handler.py @@ -0,0 +1,37 @@ +from types import SimpleNamespace + +import pytest + +from app.telegram.handlers import error_handler + + +@pytest.mark.asyncio +async def test_handle_exception_value_error_message_path(monkeypatch, fake_state, fake_message): + state = fake_state + state._data = {"messages_to_delete": [1, 2]} + msg = type(fake_message)(text="x") + update = SimpleNamespace(message=msg, callback_query=None, bot=msg.bot) + event = SimpleNamespace(update=update, exception=ValueError("bad value")) + + await error_handler.handle_exception(event, state) + + msg.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_handle_exception_validation_error_callback_truncates(monkeypatch, fake_callback): + callback = type(fake_callback)() + update = SimpleNamespace(message=None, callback_query=callback, bot=callback.bot) + + class FakeValidationError(Exception): + pass + + monkeypatch.setattr(error_handler, "ValidationError", FakeValidationError) + monkeypatch.setattr(error_handler, "format_validation_error", lambda exc: "x" * 250) + event = SimpleNamespace(update=update, exception=FakeValidationError("boom")) + + await error_handler.handle_exception(event, state=None) + + callback.answer.assert_awaited_once() + sent_text = callback.answer.await_args.args[0] + assert len(sent_text) == 200 From ca008096ede9b6e189c10a2c8af8750bd6dbf132 Mon Sep 17 00:00:00 2001 From: default Date: Wed, 22 Jul 2026 13:39:01 +0000 Subject: [PATCH 2/7] fix --- tests/api/__init__.py | 8 +++++ tests/telegram/conftest.py | 1 + .../handlers/admin/test_bulk_actions.py | 16 +++++----- tests/telegram/handlers/admin/test_user.py | 26 ++++++++-------- tests/test_record_usages.py | 30 ++++++++++++++++--- 5 files changed, 56 insertions(+), 25 deletions(-) diff --git a/tests/api/__init__.py b/tests/api/__init__.py index 71100544c..2b83afc4b 100644 --- a/tests/api/__init__.py +++ b/tests/api/__init__.py @@ -102,6 +102,14 @@ async def create_tables(): if TEST_FROM == "local": + if "test.db" in DATABASE_URL: + import os + for f in ["test.db", "test.db-shm", "test.db-wal"]: + if os.path.exists(f): + try: + os.remove(f) + except Exception: + pass run_migrations() diff --git a/tests/telegram/conftest.py b/tests/telegram/conftest.py index 76d6ecf69..0bc74bc72 100644 --- a/tests/telegram/conftest.py +++ b/tests/telegram/conftest.py @@ -57,6 +57,7 @@ def __init__(self, text: str = "text", message_id: int = 100, chat_id: int = 1): self.edit_text = AsyncMock() self.edit_reply_markup = AsyncMock() self.answer_document = AsyncMock() + self.answer_photo = AsyncMock() self.delete = AsyncMock() async def _new_message(self, *args, **kwargs): diff --git a/tests/telegram/handlers/admin/test_bulk_actions.py b/tests/telegram/handlers/admin/test_bulk_actions.py index a051418d5..03aece986 100644 --- a/tests/telegram/handlers/admin/test_bulk_actions.py +++ b/tests/telegram/handlers/admin/test_bulk_actions.py @@ -44,20 +44,20 @@ async def test_helpers_message_target_and_chunking(fake_message, fake_callback): @pytest.mark.asyncio -async def test_bulk_actions_menu(fake_message, fake_callback): +async def test_bulk_actions_menu(admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) - await bulk_h.bulk_actions(event) + await bulk_h.bulk_actions(event, admin=admin) event.message.edit_text.assert_awaited_once() @pytest.mark.asyncio -async def test_bulk_create_from_template_no_templates(monkeypatch, fake_state, fake_message, fake_callback): +async def test_bulk_create_from_template_no_templates(monkeypatch, fake_state, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) monkeypatch.setattr(bulk_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) - await bulk_h.bulk_create_from_template(event, db=object(), state=fake_state) + await bulk_h.bulk_create_from_template(event, db=object(), state=fake_state, admin=admin) event.answer.assert_awaited_once() @@ -187,12 +187,12 @@ async def test_process_expiry_invalid(fake_state, fake_message): @pytest.mark.asyncio -async def test_modify_expiry_done(monkeypatch, fake_message, fake_callback): +async def test_modify_expiry_done(monkeypatch, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) callback_data = BulkActionPanel.Callback(action=BulkAction.modify_expiry, amount="5") monkeypatch.setattr(bulk_h.user_operations, "bulk_modify_expire", AsyncMock(return_value=7)) - await bulk_h.modify_expiry_done(event, db=object(), callback_data=callback_data) + await bulk_h.modify_expiry_done(event, db=object(), admin=admin, callback_data=callback_data) event.message.edit_text.assert_awaited_once() @@ -216,11 +216,11 @@ async def test_process_data_limit_invalid(fake_state, fake_message): @pytest.mark.asyncio -async def test_modify_data_limit_done(monkeypatch, fake_message, fake_callback): +async def test_modify_data_limit_done(monkeypatch, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) callback_data = BulkActionPanel.Callback(action=BulkAction.modify_data_limit, amount="4") monkeypatch.setattr(bulk_h.user_operations, "bulk_modify_datalimit", AsyncMock(return_value=9)) - await bulk_h.modify_data_limit_done(event, db=object(), callback_data=callback_data) + await bulk_h.modify_data_limit_done(event, db=object(), admin=admin, callback_data=callback_data) event.message.edit_text.assert_awaited_once() diff --git a/tests/telegram/handlers/admin/test_user.py b/tests/telegram/handlers/admin/test_user.py index e6d0ee52e..2386e8948 100644 --- a/tests/telegram/handlers/admin/test_user.py +++ b/tests/telegram/handlers/admin/test_user.py @@ -81,46 +81,46 @@ async def test_process_data_limit_validation(fake_state, fake_message): @pytest.mark.asyncio -async def test_process_expire_zero(monkeypatch, fake_state, groups_response, fake_message): +async def test_process_expire_zero(monkeypatch, fake_state, admin, groups_response, fake_message): event = type(fake_message)(text="0") monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) monkeypatch.setattr(user_h.group_operations, "get_all_groups", AsyncMock(return_value=groups_response)) - await user_h.process_expire(event, fake_state, db=object()) + await user_h.process_expire(event, fake_state, db=object(), admin=admin) assert await fake_state.get_state() == user_h.forms.CreateUser.group_ids @pytest.mark.asyncio -async def test_process_status_on_hold(monkeypatch, fake_state, fake_message, fake_callback): +async def test_process_status_on_hold(monkeypatch, fake_state, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) callback_data = ChooseStatus.Callback(status=UserStatus.on_hold.value) monkeypatch.setattr(user_h, "delete_messages", AsyncMock()) monkeypatch.setattr(user_h, "add_to_messages_to_delete", AsyncMock()) - await user_h.process_status(event, db=object(), state=fake_state, callback_data=callback_data) + await user_h.process_status(event, db=object(), state=fake_state, callback_data=callback_data, admin=admin) assert await fake_state.get_state() == user_h.forms.CreateUser.on_hold_timeout @pytest.mark.asyncio -async def test_process_on_hold_timeout_invalid(fake_state, fake_message): +async def test_process_on_hold_timeout_invalid(fake_state, admin, fake_message): event = type(fake_message)(text="bad") - await user_h.process_on_hold_timeout(event, fake_state, db=object()) + await user_h.process_on_hold_timeout(event, fake_state, db=object(), admin=admin) event.reply.assert_awaited_once() @pytest.mark.asyncio -async def test_select_groups_toggles(monkeypatch, fake_state, groups_response, fake_message, fake_callback): +async def test_select_groups_toggles(monkeypatch, fake_state, admin, groups_response, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) callback_data = GroupsSelector.Callback(action=SelectGroupAction.select, group_id=1, user_id=0) monkeypatch.setattr(user_h.group_operations, "get_all_groups", AsyncMock(return_value=groups_response)) - await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data) - await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data) + await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data, admin=admin) + await user_h.select_groups(event, db=object(), state=fake_state, callback_data=callback_data, admin=admin) assert await fake_state.get_value("group_ids") == [] @@ -266,12 +266,12 @@ async def test_direct_user_actions(monkeypatch, admin, fake_user, func_name, act @pytest.mark.asyncio -async def test_modify_with_template_no_templates(monkeypatch, fake_message, fake_callback): +async def test_modify_with_template_no_templates(monkeypatch, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.modify_with_template) monkeypatch.setattr(user_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) - await user_h.modify_with_template(event, db=object(), callback_data=callback_data) + await user_h.modify_with_template(event, db=object(), admin=admin, callback_data=callback_data) event.answer.assert_awaited_once() @@ -290,11 +290,11 @@ async def test_modify_with_template_done(monkeypatch, admin, fake_user, fake_mes @pytest.mark.asyncio -async def test_create_user_from_template_no_templates(monkeypatch, fake_message, fake_callback): +async def test_create_user_from_template_no_templates(monkeypatch, admin, fake_message, fake_callback): event = type(fake_callback)(message=type(fake_message)()) monkeypatch.setattr(user_h.user_templates, "get_user_templates", AsyncMock(return_value=[])) - await user_h.create_user_from_template(event, db=object()) + await user_h.create_user_from_template(event, db=object(), admin=admin) event.answer.assert_awaited_once() diff --git a/tests/test_record_usages.py b/tests/test_record_usages.py index b2151bc43..46e1b1477 100644 --- a/tests/test_record_usages.py +++ b/tests/test_record_usages.py @@ -6,7 +6,7 @@ from unittest.mock import AsyncMock import pytest -from sqlalchemy import select +from sqlalchemy import select, text from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool, StaticPool @@ -61,8 +61,19 @@ async def session_factory(monkeypatch: pytest.MonkeyPatch): engine = create_async_engine(database_url, connect_args=connect_args, **engine_kwargs) async with engine.begin() as conn: - await conn.run_sync(base.Base.metadata.drop_all) - await conn.run_sync(base.Base.metadata.create_all) + if is_sqlite: + await conn.run_sync(base.Base.metadata.drop_all) + await conn.run_sync(base.Base.metadata.create_all) + else: + if database_url.startswith("postgresql"): + table_names = ", ".join(f'"{table.name}"' for table in base.Base.metadata.sorted_tables) + if table_names: + await conn.execute(text(f"TRUNCATE TABLE {table_names} CASCADE;")) + elif database_url.startswith("mysql"): + await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0;")) + for table in base.Base.metadata.sorted_tables: + await conn.execute(text(f"TRUNCATE TABLE `{table.name}`;")) + await conn.execute(text("SET FOREIGN_KEY_CHECKS = 1;")) # Seed the 3 default roles so FK constraints on admins.role_id are satisfied async with async_sessionmaker(bind=engine, expire_on_commit=False)() as seed_session: @@ -96,7 +107,18 @@ async def __aexit__(self, exc_type, exc_value, traceback): yield session_factory async with engine.begin() as conn: - await conn.run_sync(base.Base.metadata.drop_all) + if is_sqlite: + await conn.run_sync(base.Base.metadata.drop_all) + else: + if database_url.startswith("postgresql"): + table_names = ", ".join(f'"{table.name}"' for table in base.Base.metadata.sorted_tables) + if table_names: + await conn.execute(text(f"TRUNCATE TABLE {table_names} CASCADE;")) + elif database_url.startswith("mysql"): + await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0;")) + for table in base.Base.metadata.sorted_tables: + await conn.execute(text(f"TRUNCATE TABLE `{table.name}`;")) + await conn.execute(text("SET FOREIGN_KEY_CHECKS = 1;")) await engine.dispose() if needs_json_default_fix and proxy_column is not None: proxy_column.server_default = proxy_default From ce485d7e04f14ca3b1dfbb058169d47ab6f93e81 Mon Sep 17 00:00:00 2001 From: default Date: Wed, 22 Jul 2026 13:47:04 +0000 Subject: [PATCH 3/7] fix 2 --- tests/telegram/handlers/test_base.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/telegram/handlers/test_base.py b/tests/telegram/handlers/test_base.py index 5c02606f4..d4a59859e 100644 --- a/tests/telegram/handlers/test_base.py +++ b/tests/telegram/handlers/test_base.py @@ -21,6 +21,11 @@ def patch_texts(monkeypatch, fake_message, fake_callback): monkeypatch.setattr(base, "Texts", _DummyTexts()) monkeypatch.setattr(base.types, "Message", type(fake_message)) monkeypatch.setattr(base.types, "CallbackQuery", type(fake_callback)) + monkeypatch.setattr( + base, + "telegram_settings", + AsyncMock(return_value=type("S", (), {"mini_app_login": False, "mini_app_web_url": "https://panel"})()), + ) @pytest.mark.asyncio From 403e9568bd8ebe9e8d6eaad38184833659ee0dbd Mon Sep 17 00:00:00 2001 From: default Date: Wed, 22 Jul 2026 13:59:57 +0000 Subject: [PATCH 4/7] fix 3 --- tests/test_record_usages.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_record_usages.py b/tests/test_record_usages.py index 46e1b1477..4c3c80d6e 100644 --- a/tests/test_record_usages.py +++ b/tests/test_record_usages.py @@ -68,7 +68,7 @@ async def session_factory(monkeypatch: pytest.MonkeyPatch): if database_url.startswith("postgresql"): table_names = ", ".join(f'"{table.name}"' for table in base.Base.metadata.sorted_tables) if table_names: - await conn.execute(text(f"TRUNCATE TABLE {table_names} CASCADE;")) + await conn.execute(text(f"TRUNCATE TABLE {table_names} RESTART IDENTITY CASCADE;")) elif database_url.startswith("mysql"): await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0;")) for table in base.Base.metadata.sorted_tables: @@ -113,7 +113,7 @@ async def __aexit__(self, exc_type, exc_value, traceback): if database_url.startswith("postgresql"): table_names = ", ".join(f'"{table.name}"' for table in base.Base.metadata.sorted_tables) if table_names: - await conn.execute(text(f"TRUNCATE TABLE {table_names} CASCADE;")) + await conn.execute(text(f"TRUNCATE TABLE {table_names} RESTART IDENTITY CASCADE;")) elif database_url.startswith("mysql"): await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0;")) for table in base.Base.metadata.sorted_tables: From 5a3559298e66e510dd04c42c1e05a0683dc8f212 Mon Sep 17 00:00:00 2001 From: default Date: Wed, 22 Jul 2026 14:18:02 +0000 Subject: [PATCH 5/7] Delete the configured SQLite database path, not CWD test.db --- tests/api/__init__.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/tests/api/__init__.py b/tests/api/__init__.py index 2b83afc4b..60b1344a0 100644 --- a/tests/api/__init__.py +++ b/tests/api/__init__.py @@ -102,14 +102,20 @@ async def create_tables(): if TEST_FROM == "local": - if "test.db" in DATABASE_URL: + if IS_SQLITE: + from sqlalchemy.engine import make_url import os - for f in ["test.db", "test.db-shm", "test.db-wal"]: - if os.path.exists(f): - try: - os.remove(f) - except Exception: - pass + try: + db_path = make_url(DATABASE_URL).database + if db_path and db_path != ":memory:": + for f in [db_path, f"{db_path}-shm", f"{db_path}-wal"]: + if os.path.exists(f): + try: + os.remove(f) + except Exception: + pass + except Exception: + pass run_migrations() From 9e466167d4f6d0ad24ed0a6cbf6556b5a9e9d969 Mon Sep 17 00:00:00 2001 From: default Date: Wed, 22 Jul 2026 14:22:48 +0000 Subject: [PATCH 6/7] Delete the configured SQLite database path, not CWD test.db --- tests/api/__init__.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/tests/api/__init__.py b/tests/api/__init__.py index 60b1344a0..6cf9c2745 100644 --- a/tests/api/__init__.py +++ b/tests/api/__init__.py @@ -1,5 +1,6 @@ import asyncio import json +from pathlib import Path from alembic.command import upgrade from alembic.config import Config @@ -103,17 +104,13 @@ async def create_tables(): if TEST_FROM == "local": if IS_SQLITE: - from sqlalchemy.engine import make_url - import os try: - db_path = make_url(DATABASE_URL).database - if db_path and db_path != ":memory:": - for f in [db_path, f"{db_path}-shm", f"{db_path}-wal"]: - if os.path.exists(f): - try: - os.remove(f) - except Exception: - pass + database_file = Path(DATABASE_URL.partition("///")[2].split("?", 1)[0]) + for suffix in ("", "-shm", "-wal"): + try: + database_file.with_name(f"{database_file.name}{suffix}").unlink() + except FileNotFoundError: + pass except Exception: pass run_migrations() From 51e57e1471078391693db7b2a8679c0de929134d Mon Sep 17 00:00:00 2001 From: default Date: Mon, 27 Jul 2026 10:46:40 +0000 Subject: [PATCH 7/7] update telegram tests --- tests/telegram/handlers/admin/test_user.py | 35 ++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/tests/telegram/handlers/admin/test_user.py b/tests/telegram/handlers/admin/test_user.py index 2386e8948..25c7ffa90 100644 --- a/tests/telegram/handlers/admin/test_user.py +++ b/tests/telegram/handlers/admin/test_user.py @@ -389,3 +389,38 @@ async def test_search_user_and_debug(monkeypatch, admin, fake_user, fake_inline_ inline.answer.assert_awaited_once() callback.answer.assert_awaited_once_with("any:payload", show_alert=True) + + +@pytest.mark.asyncio +async def test_get_subscription_qr(monkeypatch, admin, fake_user, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.subscription_qr) + + mock_get_validated_user_by_id = AsyncMock(return_value=fake_user) + mock_validate_user = AsyncMock(return_value=fake_user) + mock_send_subscription_qr = AsyncMock() + + monkeypatch.setattr(user_h.user_operations, "get_validated_user_by_id", mock_get_validated_user_by_id) + monkeypatch.setattr(user_h.user_operations, "validate_user", mock_validate_user) + monkeypatch.setattr(user_h, "send_subscription_qr", mock_send_subscription_qr) + + await user_h.get_subscription_qr(event, db=object(), admin=admin, callback_data=callback_data) + + mock_get_validated_user_by_id.assert_awaited_once() + mock_validate_user.assert_awaited_once() + mock_send_subscription_qr.assert_awaited_once_with(event.message, fake_user.subscription_url, fake_user.username) + event.answer.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_subscription_qr_not_found(monkeypatch, admin, fake_message, fake_callback): + event = type(fake_callback)(message=type(fake_message)()) + callback_data = UserPanel.Callback(user_id=11, action=UserPanelAction.subscription_qr) + + mock_get_validated_user_by_id = AsyncMock(side_effect=ValueError("user not found")) + monkeypatch.setattr(user_h.user_operations, "get_validated_user_by_id", mock_get_validated_user_by_id) + + await user_h.get_subscription_qr(event, db=object(), admin=admin, callback_data=callback_data) + + event.answer.assert_awaited_once_with(user_h.Texts.user_not_found, show_alert=True) +