tests/test_adapter.py - master - rustwizard/max_messenger_adapter

tests/test_adapter.py
217 строк8 KB
Rust Wizard
test: extract shared Hermes gateway mocks into conftest.py
20 июн 2026, 10:57
20 июн 2026, 10:57370c435
"""Tests for the MAX Hermes platform adapter."""

from __future__ import annotations

import hashlib
import hmac
import os
from unittest.mock import MagicMock, patch

import pytest

from conftest import (
    FakePlatformConfig,
    FakePluginContext,
)

import adapter as max_adapter


class TestHelpers:
    def test_parse_chat_id_user_prefix(self):
        assert max_adapter._parse_chat_id("u:123") == (None, 123)

    def test_parse_chat_id_chat_prefix(self):
        assert max_adapter._parse_chat_id("c:456") == (456, None)

    def test_parse_chat_id_plain_number(self):
        assert max_adapter._parse_chat_id("789") == (789, None)

    def test_parse_chat_id_invalid(self):
        assert max_adapter._parse_chat_id("not-a-number") == (None, None)

    def test_split_text_short(self):
        assert max_adapter.MaxAdapter._split_text("hello", 10) == ["hello"]

    def test_split_text_long_prefers_space(self):
        text = "hello world " * 1000
        chunks = max_adapter.MaxAdapter._split_text(text, 100)
        assert all(len(c) <= 100 for c in chunks)

    def test_verify_webhook_signature_no_secret(self):
        assert max_adapter._verify_webhook_signature(b"body", "sig", "") is True

    def test_verify_webhook_signature_missing_sig(self):
        assert max_adapter._verify_webhook_signature(b"body", "", "secret") is False

    def test_verify_webhook_signature_valid(self):
        secret = "secret"
        body = b"body"
        sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        assert max_adapter._verify_webhook_signature(body, sig, secret) is True


class TestConfig:
    def test_validate_config_missing_token(self):
        cfg = FakePlatformConfig()
        with patch.dict(os.environ, {"MAX_BOT_TOKEN": ""}, clear=True):
            assert max_adapter.validate_config(cfg) is False

    def test_validate_config_with_env_token(self):
        cfg = FakePlatformConfig()
        with patch.dict(os.environ, {"MAX_BOT_TOKEN": "tok"}, clear=True):
            assert max_adapter.validate_config(cfg) is True

    def test_validate_config_with_extra_token(self):
        cfg = FakePlatformConfig(extra={"token": "tok"})
        with patch.dict(os.environ, {"MAX_BOT_TOKEN": ""}, clear=True):
            assert max_adapter.validate_config(cfg) is True

    def test_env_enablement_returns_none_without_token(self):
        with patch.dict(os.environ, {"MAX_BOT_TOKEN": ""}, clear=True):
            assert max_adapter._env_enablement() is None

    def test_env_enablement_populates_seed(self):
        env = {
            "MAX_BOT_TOKEN": "tok",
            "MAX_API_BASE_URL": "https://api.max.ru",
            "MAX_WEBHOOK_URL": "https://example.com/webhook",
            "MAX_HOME_CHANNEL": "123",
        }
        with patch.dict(os.environ, env, clear=True):
            seed = max_adapter._env_enablement()
        assert seed is not None
        assert seed["token"] == "tok"
        assert seed["api_base_url"] == "https://api.max.ru"
        assert seed["webhook_url"] == "https://example.com/webhook"
        assert seed["home_channel"] == {"chat_id": "123", "name": "123"}


class TestRegistration:
    def test_register(self):
        ctx = FakePluginContext()
        max_adapter.register(ctx)
        assert "max" in ctx.entries
        entry = ctx.entries["max"]
        assert entry.label == "MAX"


class TestWebhookServer:
    @pytest.mark.asyncio
    async def test_health_endpoint(self, aiohttp_client):
        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter._app = max_adapter.web.Application()
        adapter._app.router.add_get("/health", adapter._handle_health_request)
        client = await aiohttp_client(adapter._app)
        resp = await client.get("/health")
        assert resp.status == 200
        body = await resp.json()
        assert body["status"] == "ok"
        await client.close()


class TestStateMaps:
    def test_approval_state_stored(self):
        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter._approval_state[1] = "session:key"
        assert adapter._approval_state[1] == "session:key"

    def test_slash_confirm_state_stored(self):
        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter._slash_confirm_state["abc"] = "session:key"
        assert adapter._slash_confirm_state["abc"] == "session:key"

    def test_clarify_state_stored(self):
        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter._clarify_state["cl:1"] = "session:key"
        assert adapter._clarify_state["cl:1"] == "session:key"


class TestResolveTarget:
    def test_resolve_user_prefix(self):
        adapter = max_adapter.MaxAdapter(FakePlatformConfig(extra={"token": "tok"}))
        assert max_adapter.MaxAdapter._resolve_target(adapter, "u:123") == (None, 123)

    def test_resolve_chat_prefix(self):
        adapter = max_adapter.MaxAdapter(FakePlatformConfig(extra={"token": "tok"}))
        assert max_adapter.MaxAdapter._resolve_target(adapter, "c:456") == (456, None)

    def test_resolve_known_dm(self):
        adapter = max_adapter.MaxAdapter(FakePlatformConfig(extra={"token": "tok"}))
        adapter._known_chats = {"42": {"type": "dm", "chat_id": None, "user_id": 42}}
        assert max_adapter.MaxAdapter._resolve_target(adapter, "42") == (None, 42)

    def test_resolve_known_group(self):
        adapter = max_adapter.MaxAdapter(FakePlatformConfig(extra={"token": "tok"}))
        adapter._known_chats = {"99": {"type": "group", "chat_id": 99}}
        assert max_adapter.MaxAdapter._resolve_target(adapter, "99") == (99, None)


@pytest.mark.skipif(
    not max_adapter.MAXBOT_AVAILABLE, reason="maxbot-api-client-python not installed"
)
class TestMessageEventBuilding:
    def test_build_dm_message_event(self):
        from maxbot_api_client_python.types.constants import ChatType
        from maxbot_api_client_python.types.models import (
            Message,
            MessageBody,
            Recipient,
            User,
        )

        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter.build_source = MagicMock(return_value=MagicMock())

        msg = Message(
            sender=User(
                user_id=42,
                first_name="Ivan",
                last_name="Ivanov",
                username="ivan",
                is_bot=False,
                last_activity_time=0,
            ),
            recipient=Recipient(chat_id=42, chat_type=ChatType.DIALOG, user_id=42),
            body=MessageBody(mid="mid:123", seq=1, text="Hello"),
            timestamp=1234567890,
        )
        event = adapter._build_message_event(msg, max_adapter.MessageType.TEXT)
        assert event is not None
        assert event.text == "Hello"
        assert event.message_type == max_adapter.MessageType.TEXT
        assert adapter._known_chats["42"]["type"] == "dm"

    def test_build_group_message_event(self):
        from maxbot_api_client_python.types.constants import ChatType
        from maxbot_api_client_python.types.models import (
            Message,
            MessageBody,
            Recipient,
            User,
        )

        cfg = FakePlatformConfig(extra={"token": "tok"})
        adapter = max_adapter.MaxAdapter(cfg)
        adapter.build_source = MagicMock(return_value=MagicMock())

        msg = Message(
            sender=User(
                user_id=42,
                first_name="Ivan",
                is_bot=False,
                last_activity_time=0,
            ),
            recipient=Recipient(chat_id=99, chat_type=ChatType.CHAT),
            body=MessageBody(mid="mid:456", seq=1, text="Hello group"),
            timestamp=1234567890,
        )
        event = adapter._build_message_event(msg, max_adapter.MessageType.TEXT)
        assert event is not None
        assert event.text == "Hello group"
        assert adapter._known_chats["99"]["type"] == "group"