hermes-chatto-plugin/test_adapter.py

"""Unit tests for the Chatto platform adapter.

Covers:
  - Emoji shortcode conversion
  - Adapter instantiation and properties
  - Registration and requirements
  - Basic functionality with chattolib
  - Message sending and reactions
  - User lookup (with caching)
  - Presence and custom status

All network calls are mocked — no real HTTP or WebSocket connections.
"""

import asyncio
import os
import sys
import tempfile
from unittest.mock import AsyncMock, MagicMock, patch, call
from collections import OrderedDict

import pytest
import pytest_asyncio

# Import chattolib types for tests - using vendored chattolib from adapter

# -- Path setup --
PLUGIN_ROOT = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, PLUGIN_ROOT)
sys.path.insert(0, "/opt/hermes")
sys.path.insert(0, "/root/.hermes/plugins/platforms/chatto")

from adapter import (
    ChattoAdapter,
    hermes_check_fn as check_requirements,
    hermes_validate_config as validate_config,
    register,
)
from platform_config import ChattoConstants
from gateway.config import PlatformConfig
from gateway.platforms.base import SendResult, MessageEvent, MessageType

_EMOJI_TO_SHORTCODE = ChattoConstants.EMOJI_TO_SHORTCODE
_MAX_MESSAGE_LENGTH = ChattoConstants.MAX_MESSAGE_LENGTH
_SEEN_CAP = ChattoConstants.SEEN_CAP


# -- Helpers --

class _MockPluginContext:
    """Minimal mock for the plugin registration context."""

    def __init__(self):
        self.registered_names = []
        self.registered_kwargs = None

    def register_platform(self, **kwargs):
        from gateway.platform_registry import platform_registry, PlatformEntry
        entry = PlatformEntry(
            name=kwargs["name"],
            label=kwargs.get("label", kwargs["name"]),
            adapter_factory=kwargs.get("adapter_factory"),
            check_fn=kwargs.get("check_fn"),
            validate_config=kwargs.get("validate_config"),
            is_connected=kwargs.get("is_connected"),
            required_env=kwargs.get("required_env", []),
            source="plugin",
        )
        platform_registry.register(entry)
        self.registered_names.append(kwargs["name"])
        self.registered_kwargs = kwargs


def _ensure_chatto_registered():
    """Register chatto in the platform registry so Platform('chatto') works."""
    from gateway.platform_registry import platform_registry
    if not platform_registry.is_registered("chatto"):
        ctx = _MockPluginContext()
        register(ctx)


_CHATTO_ENV_KEYS = [
    "CHATTO_BASE_URL", "CHATTO_LOGIN", "CHATTO_PASSWORD",
    "CHATTO_CHANNELS", "CHATTO_HOME_CHANNEL",
    "CHATTO_REQUIRE_MENTION", "CHATTO_ALLOWED_USERS",
    "CHATTO_ALLOW_ALL_USERS", "CHATTO_AUTO_THREAD",
    "CHATTO_REACTIONS",
]


def _clear_chatto_env(monkeypatch=None):
    """Remove all CHATTO_* env vars so tests start from a clean slate."""
    for key in _CHATTO_ENV_KEYS:
        if monkeypatch is not None:
            monkeypatch.delenv(key, raising=False)
        else:
            os.environ.pop(key, None)


def _make_config(**extra_overrides):
    """Create a minimal PlatformConfig for testing."""
    _ensure_chatto_registered()
    extra = {"base_url": "https://chat.example.com", "channels": ["room1"]}
    extra.update(extra_overrides)
    return PlatformConfig(enabled=True, extra=extra)


def _make_adapter(**extra_overrides):
    """Create a ChattoAdapter with mocked config."""
    _clear_chatto_env()
    cfg = _make_config(**extra_overrides)
    adapter = ChattoAdapter(cfg)
    adapter._chatto_client = MagicMock()
    adapter._token = "test-token"
    adapter._user_id = "bot-user-id"
    adapter._user_login = "hermes_bot"
    adapter._user_display = "Hermes Bot"
    return adapter


# -- Emoji shortcode conversion --

class TestEmojiShortcode:
    """Test emoji to shortcode mapping."""

    def test_emoji_to_shortcode_exists(self):
        assert isinstance(_EMOJI_TO_SHORTCODE, dict)
        assert len(_EMOJI_TO_SHORTCODE) > 0

    def test_emoji_to_shortcode_common_emojis(self):
        assert _EMOJI_TO_SHORTCODE.get("👍") == "thumbsup"
        assert _EMOJI_TO_SHORTCODE.get("👎") == "thumbsdown"
        assert _EMOJI_TO_SHORTCODE.get("❤️") == "heart"
        assert _EMOJI_TO_SHORTCODE.get("❤") == "heart"
        assert _EMOJI_TO_SHORTCODE.get("✅") == "white_check_mark"
        assert _EMOJI_TO_SHORTCODE.get("❌") == "x"


# -- Adapter instantiation and properties --

class TestAdapterInstantiation:
    """Test ChattoAdapter creation and basic properties."""

    def test_adapter_creation(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter is not None
        assert adapter.platform.name == "chatto"

    def test_adapter_max_message_length(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.MAX_MESSAGE_LENGTH == _MAX_MESSAGE_LENGTH

    def test_adapter_splits_long_messages(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.splits_long_messages is True

    def test_adapter_supports_threads(self):
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.supports_threads() is True


# -- Registration and requirements --

class TestRegistration:
    """Test plugin registration."""

    def test_register_called(self):
        ctx = _MockPluginContext()
        register(ctx)
        assert "chatto" in ctx.registered_names
        assert ctx.registered_kwargs["name"] == "chatto"
        assert ctx.registered_kwargs["label"] == "Chatto"

    def test_check_requirements(self):
        assert check_requirements() is True

    def test_check_requirements_missing(self):
        with patch("builtins.__import__", side_effect=ImportError("no chattolib")):
            assert check_requirements() is False

    def test_validate_config(self):
        _clear_chatto_env()
        os.environ["CHATTO_BASE_URL"] = "https://chat.test"
        os.environ["CHATTO_LOGIN"] = "user"
        os.environ["CHATTO_PASSWORD"] = "pass"
        cfg = PlatformConfig(enabled=True, extra={"base_url": "https://chat.test"})
        assert validate_config(cfg) is True
        _clear_chatto_env()


# -- Send functionality --

class TestSend:
    """Test message sending functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.post_message = AsyncMock()
        adapter._token = "test-token"
        adapter._user_id = "bot-user-id"
        return adapter

    async def test_send_calls_post_message(self, adapter):
        mock_msg = MagicMock()
        mock_msg.id = "msg-123"
        adapter._chatto_client.post_message.return_value = mock_msg
        result = await adapter.send("room-1", "Hello world")
        assert result.success is True
        assert result.message_id == "msg-123"
        adapter._chatto_client.post_message.assert_called_once()

    async def test_send_with_thread(self, adapter):
        mock_msg = MagicMock()
        mock_msg.id = "msg-456"
        adapter._chatto_client.post_message.return_value = mock_msg
        result = await adapter.send("room-1", "Hello", reply_to="thread-123")
        assert result.success is True
        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
        assert call_kwargs["thread_root_event_id"] == "thread-123"


# -- Reactions --

class TestReactions:
    """Test reaction functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.add_reaction = AsyncMock()
        adapter._chatto_client.remove_reaction = AsyncMock()
        adapter._token = "test-token"
        return adapter

    async def test_send_reaction(self, adapter):
        await adapter.add_reaction("room-1", "msg-1", "👍")
        adapter._chatto_client.add_reaction.assert_called_once()

    async def test_remove_reaction(self, adapter):
        await adapter.remove_reaction("room-1", "msg-1", "👍")
        adapter._chatto_client.remove_reaction.assert_called_once()

    async def test_on_processing_start_adds_eyes_reaction(self, adapter):
        """on_processing_start should call add_reaction with 👀."""
        event = MagicMock()
        event.message_id = "msg-1"
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_called_once()
        call_kwargs = adapter._chatto_client.add_reaction.call_args.kwargs
        assert call_kwargs["message_event_id"] == "msg-1"
        assert call_kwargs["room_id"] == "room-1"
        assert call_kwargs["emoji"] == "eyes"

    async def test_on_processing_start_empty_message_id(self, adapter):
        """on_processing_start should skip reaction when message_id is empty."""
        event = MagicMock()
        event.message_id = None
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_not_called()

    async def test_on_processing_start_reactions_disabled(self, adapter):
        """on_processing_start should skip when reactions config is False."""
        adapter.chatto_config.reactions.value = False
        event = MagicMock()
        event.message_id = "msg-1"
        event.source.chat_id = "room-1"
        await adapter.on_processing_start(event)
        adapter._chatto_client.add_reaction.assert_not_called()


# -- Edit and Delete Messages --

class TestMessageEditing:
    """Test message editing and deletion."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.update_message = AsyncMock()
        adapter._chatto_client.delete_message = AsyncMock(return_value=True)
        adapter._token = "test-token"
        return adapter

    async def test_edit_message(self, adapter):
        result = await adapter.edit_message("room-1", "msg-1", "New content")
        assert result is True
        adapter._chatto_client.update_message.assert_called_once()

    async def test_delete_message(self, adapter):
        result = await adapter.delete_message("room-1", "msg-1")
        assert result is True
        adapter._chatto_client.delete_message.assert_called_once()


# -- User lookup --

class TestUserLookup:
    """Test user lookup functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._token = "test-token"
        adapter._user_cache = {}
        return adapter

    async def test_get_user_calls_chattolib(self, adapter):
        # Import chattolib types for testing (try vendored first)
        try:
            from chattolib_vendor.chattolib.types import User, GetUserResponse
        except ImportError:
            try:
                from chattolib.types import User, GetUserResponse
            except ImportError:
                # Fallback to mocked types if chattolib not installed
                from unittest.mock import MagicMock
                User = MagicMock
                GetUserResponse = MagicMock
        mock_user = User(id="user-1", login="testuser", display_name="Test User")
        adapter._chatto_client.get_user.return_value = GetUserResponse(user=mock_user)
        result = await adapter.get_user("user-1")
        assert result is not None
        assert result["id"] == "user-1"
        assert result["login"] == "testuser"

    async def test_get_user_caching(self, adapter):
        # Import chattolib types for testing (try vendored first)
        try:
            from chattolib_vendor.chattolib.types import User, GetUserResponse
        except ImportError:
            try:
                from chattolib.types import User, GetUserResponse
            except ImportError:
                # Fallback to mocked types if chattolib not installed
                from unittest.mock import MagicMock
                User = MagicMock
                GetUserResponse = MagicMock
        mock_user = User(id="user-1", login="testuser", display_name="Test User")
        adapter._chatto_client.get_user.return_value = GetUserResponse(user=mock_user)
        result1 = await adapter.get_user("user-1")
        result2 = await adapter.get_user("user-1")
        assert result1 == result2
        assert adapter._chatto_client.get_user.call_count == 1


# -- Presence and Custom Status --

class TestPresence:
    """Test presence functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.update_presence = AsyncMock()
        adapter._token = "test-token"
        return adapter

    async def test_set_presence(self, adapter):
        result = await adapter.set_presence("online")
        assert result is True
        adapter._chatto_client.update_presence.assert_called_once()


class TestCustomStatus:
    """Test custom status functionality."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._chatto_client.update_custom_status = AsyncMock()
        adapter._chatto_client.delete_custom_status = AsyncMock()
        adapter._token = "test-token"
        return adapter

    async def test_set_custom_status(self, adapter):
        result = await adapter.set_custom_status("Processing...")
        assert result is True
        adapter._chatto_client.update_custom_status.assert_called_once()

    async def test_clear_custom_status(self, adapter):
        result = await adapter.clear_custom_status()
        assert result is True
        adapter._chatto_client.delete_custom_status.assert_called_once()


# -- Room operations --

class TestRoomOperations:
    """Test room creation and DM initiation."""

    @pytest_asyncio.fixture
    def adapter(self):
        _clear_chatto_env()
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        adapter._chatto_client = MagicMock()
        adapter._token = "test-token"
        adapter._room_names = {}
        adapter._room_kinds = {}
        return adapter

    async def test_create_room(self, adapter):
        # Import chattolib types for testing (try vendored first)
        try:
            from chattolib_vendor.chattolib.types import Room
        except ImportError:
            try:
                from chattolib.types import Room
            except ImportError:
                from unittest.mock import MagicMock
                Room = MagicMock
        mock_room = Room(id="room-123", name="Test Room", kind="ROOM_KIND_GROUP",
                        description="", archived=False, group_id="", universal=True)
        adapter._chatto_client.create_room.return_value = mock_room
        result = await adapter.create_room("Test Room", "A test room")
        assert result == "room-123"
        adapter._chatto_client.create_room.assert_called_once()

    async def test_start_dm(self, adapter):
        # Import chattolib types for testing (try vendored first)
        try:
            from chattolib_vendor.chattolib.types import Room
        except ImportError:
            try:
                from chattolib.types import Room
            except ImportError:
                from unittest.mock import MagicMock
                Room = MagicMock
        mock_room = Room(id="dm-123", name="DM with user", kind="ROOM_KIND_DM",
                        description="", archived=False, group_id="", universal=False)
        adapter._chatto_client.start_dm.return_value = mock_room
        result = await adapter.start_dm("user-123")
        assert result == "dm-123"
        adapter._chatto_client.start_dm.assert_called_once()


# -- Constants --

class TestConstants:
    """Test that constants are properly defined."""

    def test_max_message_length(self):
        assert _MAX_MESSAGE_LENGTH == 10000

    def test_seen_cap(self):
        assert _SEEN_CAP == 500