TelegramBotAdmin/tests/conftest.py

from __future__ import annotations

import importlib
import sys
import types
from pathlib import Path
from types import SimpleNamespace

import pytest
from mongomock_motor import AsyncMongoMockClient
from pyrogram import filters
from pyrogram.enums import ChatMemberStatus

ROOT = Path(__file__).resolve().parents[1]


class FakeTelegramApp:
    def __init__(self) -> None:
        self.members: dict[tuple[int, int], object] = {}
        self.chats: dict[int, object] = {}
        self.histories: dict[int, list[object]] = {}
        self.sent_messages: list[tuple[int, str]] = []
        self.sent_photos: list[tuple[int, object, str]] = []
        self.edited_messages: list[tuple[int, int, str]] = []
        self.pinned_messages: list[tuple[int, int]] = []
        self.unpinned_messages: list[tuple[int, int]] = []
        self.deleted_messages: list[tuple[int, int]] = []

    async def get_chat_member(self, chat_id: int, user_id: int):
        member = self.members.get((int(chat_id), int(user_id)))
        if member is not None:
            return member
        return SimpleNamespace(
            status=ChatMemberStatus.MEMBER,
            privileges=None,
            user=SimpleNamespace(
                id=int(user_id),
                username=None,
                first_name=str(user_id),
                last_name=None,
                is_bot=False,
                is_deleted=False,
            ),
            custom_title=None,
            until_date=None,
        )

    async def get_chat(self, chat_id: int):
        chat = self.chats.get(int(chat_id))
        if chat is None:
            raise ValueError("chat not found")
        return chat

    async def get_chat_history(self, chat_id: int, limit: int = 100):
        for message in self.histories.get(int(chat_id), [])[:limit]:
            yield message

    async def send_message(self, chat_id: int, text: str, **_kwargs):
        self.sent_messages.append((int(chat_id), text))
        return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=len(self.sent_messages))

    async def send_photo(self, chat_id: int, photo: object, caption: str = "", **_kwargs):
        self.sent_photos.append((int(chat_id), photo, caption))
        return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=len(self.sent_photos))

    async def edit_message_text(
        self, chat_id: int, message_id: int, text: str, **_kwargs
    ):
        self.edited_messages.append((int(chat_id), int(message_id), text))
        return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=int(message_id))

    async def pin_chat_message(
        self, chat_id: int, message_id: int, **_kwargs
    ) -> None:
        self.pinned_messages.append((int(chat_id), int(message_id)))

    async def unpin_chat_message(self, chat_id: int, message_id: int) -> None:
        self.unpinned_messages.append((int(chat_id), int(message_id)))

    async def delete_messages(self, chat_id: int, message_id: int) -> None:
        self.deleted_messages.append((int(chat_id), int(message_id)))

    def on_message(self, *_args, **_kwargs):
        return lambda function: function

    def on_callback_query(self, *_args, **_kwargs):
        return lambda function: function

    def on_edited_message(self, *_args, **_kwargs):
        return lambda function: function

    def on_deleted_messages(self, *_args, **_kwargs):
        return lambda function: function

    def on_chat_member_updated(self, *_args, **_kwargs):
        return lambda function: function


@pytest.fixture
def app_modules(tmp_path):
    previous = {
        name: module
        for name, module in sys.modules.items()
        if name == "wbb" or name.startswith("wbb.")
    }
    for name in list(previous):
        sys.modules.pop(name, None)

    fake_app = FakeTelegramApp()
    fake_wbb = types.ModuleType("wbb")
    fake_wbb.__path__ = [str(ROOT / "wbb")]
    fake_wbb.db = AsyncMongoMockClient().wbb
    fake_wbb.control_db = fake_wbb.db
    fake_wbb.app = fake_app
    fake_wbb.aiohttpsession = None
    fake_wbb.log = SimpleNamespace(info=lambda _message: None, error=lambda _message: None)
    fake_wbb.BOT_ID = 999
    fake_wbb.BOT_NAME = "Test Bot"
    fake_wbb.BOT_USERNAME = "test_bot"
    fake_wbb.BOT_PROFILE_ID = "primary"
    fake_wbb.BOT_PERMISSIONS = {"*"}
    fake_wbb.LOG_GROUP_ID = 0
    fake_wbb.TELEGRAM_CONNECTED = True
    fake_wbb.SUPERVISOR_MODE = False
    fake_wbb.ADMIN_BOOTSTRAP_MODE = False
    fake_wbb.BOT_SUPERVISOR = None
    fake_wbb.SUDOERS = filters.user()
    fake_wbb.MOD_LOAD = []
    fake_wbb.MOD_NOLOAD = []
    fake_wbb.MESSAGE_DUMP_CHAT = 0
    fake_wbb.ADMIN_WEB_USERNAME = "admin"
    fake_wbb.ADMIN_WEB_INITIAL_PASSWORD = "qwe0.123456"
    fake_wbb.ADMIN_WEB_SESSION_HOURS = 12
    fake_wbb.ADMIN_WEB_COOKIE_SECURE = False
    fake_wbb.ADMIN_WEB_UPLOAD_MAX_MB = 20
    fake_wbb.ADMIN_WEB_HOST = "127.0.0.1"
    fake_wbb.ADMIN_WEB_PORT = 8088
    fake_wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = (
        "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="
    )
    fake_wbb.SERVICE_ADDRESS_RETENTION_DAYS = 7
    fake_wbb.SERVICE_REVIEW_QR_EXPIRY_HOURS = 24
    fake_wbb.SERVICE_CUSTOMER_MAX_OPEN_ORDERS = 3
    fake_wbb.SERVICE_CUSTOMER_DAILY_REQUEST_LIMIT = 10
    fake_wbb.BOT_PROFILES_PATH = str(tmp_path / "bot_profiles.json")
    sys.modules["wbb"] = fake_wbb

    def load(name: str):
        return importlib.import_module(name)

    yield SimpleNamespace(wbb=fake_wbb, app=fake_app, load=load)

    for name in list(sys.modules):
        if name == "wbb" or name.startswith("wbb."):
            sys.modules.pop(name, None)
    sys.modules.update(previous)