tests/conftest.py - master - rustwizard/max_messenger_adapter

tests/conftest.py
215 строк6 KB
Rust Wizard
test: extract shared Hermes gateway mocks into conftest.py
20 июн 2026, 10:57
20 июн 2026, 10:57370c435
"""Shared test fixtures and Hermes gateway mocks.

This module is imported automatically by pytest. It injects lightweight fakes
for the Hermes gateway modules before any test imports the MAX adapter.
"""

from __future__ import annotations

import os
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
from unittest.mock import AsyncMock, MagicMock

import pytest


# ---------------------------------------------------------------------------
# Fake Hermes gateway types
# ---------------------------------------------------------------------------


@dataclass
class FakeSource:
    chat_id: str = ""
    chat_name: str = ""
    chat_type: str = ""
    user_id: str = ""
    user_name: str = ""
    message_id: str = ""


@dataclass
class FakeMessageEvent:
    text: str = ""
    message_type: Any = None
    source: Any = None
    raw_message: Any = None
    message_id: str = ""
    media_urls: List[str] = field(default_factory=list)
    media_types: List[str] = field(default_factory=list)
    reply_to_message_id: str = ""
    timestamp: int = 0


class FakeMessageType:
    TEXT = MagicMock()
    COMMAND = MagicMock()


@dataclass
class FakeSendResult:
    success: bool = True
    message_id: Optional[str] = None
    error: Optional[str] = None
    retryable: bool = False
    continuation_message_ids: Tuple[str, ...] = ()


class FakeBaseAdapter:
    """Minimal stand-in for gateway.platforms.base.BasePlatformAdapter."""

    def __init__(self, config, platform):
        self.config = config
        self.platform = platform
        self.handled_events: List[FakeMessageEvent] = []
        self._fatal_errors: List[Dict[str, Any]] = []
        self._connected = False

    def build_source(self, **kwargs):
        return FakeSource(**kwargs)

    def handle_message(self, event):
        self.handled_events.append(event)

    def _set_fatal_error(
        self, code: str, message: str, retryable: bool = False
    ) -> None:
        self._fatal_errors.append(
            {"code": code, "message": message, "retryable": retryable}
        )

    def _mark_connected(self) -> None:
        self._connected = True

    def _mark_disconnected(self) -> None:
        self._connected = False


@dataclass
class FakePlatformConfig:
    """Stand-in for gateway.config.PlatformConfig."""

    extra: Dict[str, Any] = field(default_factory=dict)
    enabled: bool = True


@dataclass
class PlatformEntry:
    """Captured register_platform() call."""

    name: str
    label: str
    adapter_factory: Any
    check_fn: Any
    validate_config: Any
    is_connected: Any
    required_env: List[str]
    install_hint: str
    env_enablement_fn: Any
    cron_deliver_env_var: str
    allowed_users_env: str
    allow_all_env: str
    max_message_length: int
    emoji: str
    platform_hint: str


class FakePluginContext:
    """Stand-in for the Hermes plugin context passed to register()."""

    def __init__(self):
        self.entries: Dict[str, PlatformEntry] = {}

    def register_platform(self, **kwargs):
        entry = PlatformEntry(**kwargs)
        self.entries[entry.name] = entry


class FakeMaxClientWrapper:
    """Fake MAX Bot API client wrapper used by MaxAdapter._client."""

    def __init__(self, token: str = "", base_url: str = ""):
        self.token = token
        self.base_url = base_url
        self.subscribe = AsyncMock(return_value=None)
        self.unsubscribe = AsyncMock(return_value=None)
        self.close = AsyncMock(return_value=None)
        self.get_bot_info = AsyncMock(
            return_value={"name": "FakeBot", "username": "fakebot"}
        )
        self.send_message = AsyncMock(return_value="mid:reply")
        self.send_chat_action = AsyncMock(return_value=None)
        self.edit_message = AsyncMock(return_value=None)
        self.delete_message = AsyncMock(return_value=None)
        self.answer_callback = AsyncMock(return_value=None)
        self.get_updates = AsyncMock(return_value=([], None))
        self.upload_file = AsyncMock(return_value="upload-token")
        self.upload_image = AsyncMock(return_value="upload-token")


# ---------------------------------------------------------------------------
# Inject gateway module mocks before adapter is imported anywhere
# ---------------------------------------------------------------------------


def _install_gateway_mocks():
    platform_mock = MagicMock()
    platform_mock.Platform = MagicMock()
    platform_mock.PlatformConfig = MagicMock()

    base_mock = MagicMock()
    base_mock.BasePlatformAdapter = FakeBaseAdapter
    base_mock.MessageEvent = FakeMessageEvent
    base_mock.MessageType = FakeMessageType
    base_mock.SendResult = FakeSendResult

    sys.modules["gateway.config"] = platform_mock
    sys.modules["gateway.platforms.base"] = base_mock


_install_gateway_mocks()

# Ensure the plugin directory is on sys.path so tests can import the adapter.
_PLUGIN_DIR = os.path.join(
    os.path.dirname(__file__), "..", "plugins", "platforms", "max"
)
if _PLUGIN_DIR not in sys.path:
    sys.path.insert(0, _PLUGIN_DIR)


# ---------------------------------------------------------------------------
# Pytest fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def plugin_ctx() -> FakePluginContext:
    return FakePluginContext()


@pytest.fixture
def platform_config() -> FakePlatformConfig:
    return FakePlatformConfig(
        extra={
            "token": "fake-token",
            "api_base_url": "https://platform-api.max.ru",
            "webhook_url": "https://example.com/webhook",
            "webhook_secret": "shhh",
            "webhook_port": 0,
        }
    )


@pytest.fixture
def fake_max_client() -> FakeMaxClientWrapper:
    return FakeMaxClientWrapper()


@pytest.fixture
def adapter_module():
    """Import and return the adapter module with gateway mocks in place."""
    import adapter as _adapter

    return _adapter