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 --
# The Hermes agent itself is not a dependency of this plugin; point HERMES_ROOT
# at a checkout to run these tests outside a deployed agent.
PLUGIN_ROOT = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, PLUGIN_ROOT)
sys.path.insert(0, os.environ.get("HERMES_ROOT", "/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 chattolib.types import Room, RoomKind
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 the platform so Platform(PLATFORM_NAME) resolves."""
    from gateway.platform_registry import platform_registry
    if not platform_registry.is_registered(ChattoConstants.PLATFORM_NAME):
        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_room(room_id, name, kind):
    """Build a real chattolib Room, as the client would return."""
    return Room(id=room_id, name=name, kind=kind, description="",
                archived=False, group_id="", universal=kind != RoomKind.DM)


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
        # Platform members created dynamically from a plugin name carry the
        # name upper-cased; the registered identity is the value.
        assert adapter.platform.value == ChattoConstants.PLATFORM_NAME

    def test_adapter_max_message_length(self):
        """The framework chunks via max_message_length_for_chat(), which reads
        the adapter-scalar MAX_MESSAGE_LENGTH and silently falls back to 4096
        when it is missing."""
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.MAX_MESSAGE_LENGTH == _MAX_MESSAGE_LENGTH
        assert adapter.max_message_length_for_chat("room-1") == _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_threads_enabled_by_default(self):
        """There is no capability flag for threads — Chatto threading is driven
        by the auto_thread setting, which defaults to on."""
        cfg = _make_config()
        adapter = ChattoAdapter(cfg)
        assert adapter.chatto_config.auto_thread.value is True


# -- Registration and requirements --

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

    def test_register_called(self):
        ctx = _MockPluginContext()
        register(ctx)
        assert ChattoConstants.PLATFORM_NAME in ctx.registered_names
        assert ctx.registered_kwargs["name"] == ChattoConstants.PLATFORM_NAME
        assert ctx.registered_kwargs["label"] == ChattoConstants.PLATFORM_LABEL
        assert ctx.registered_kwargs["max_message_length"] == _MAX_MESSAGE_LENGTH

    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

    @pytest.mark.xfail(
        strict=True,
        reason="ChattoAdapter does not override edit_message yet, so the base "
               "class reports 'Not supported' and callers send a new message "
               "instead of editing. chattolib.update_message() exists — drop "
               "this marker once the override lands.",
    )
    async def test_edit_message(self, adapter):
        result = await adapter.edit_message("room-1", "msg-1", "New content")
        assert result.success is True
        adapter._chatto_client.update_message.assert_called_once()

    @pytest.mark.xfail(
        strict=True,
        reason="ChattoAdapter does not override delete_message yet, so the base "
               "class returns False. chattolib.delete_message() exists — drop "
               "this marker once the override lands.",
    )
    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()


# NOTE: there are deliberately no tests for get_user(), set_presence() or
# set_custom_status() on the adapter. Those are not adapter responsibilities —
# callers use the chattolib client directly, which exposes them (client.get_user,
# client.update_presence, client.update_custom_status). The adapter only touches
# presence in connect()/disconnect().


# -- 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()
        # AsyncMock, not MagicMock: the adapter awaits these, and awaiting a
        # plain MagicMock raises TypeError, which create_room()/start_dm()
        # swallow into a None return.
        adapter._chatto_client.create_room = AsyncMock()
        adapter._chatto_client.start_dm = AsyncMock()
        adapter._token = "test-token"
        adapter._room_names = {}
        adapter._room_kinds = {}
        return adapter

    async def test_create_room(self, adapter):
        adapter._chatto_client.create_room.return_value = _make_room(
            "room-123", "Test Room", RoomKind.CHANNEL,
        )
        result = await adapter.create_room("Test Room", "A test room")
        assert result == "room-123"
        adapter._chatto_client.create_room.assert_called_once()
        assert adapter._room_names["room-123"] == "Test Room"

    async def test_start_dm(self, adapter):
        adapter._chatto_client.start_dm.return_value = _make_room(
            "dm-123", "DM with user", RoomKind.DM,
        )
        result = await adapter.start_dm("user-123")
        assert result == "dm-123"
        adapter._chatto_client.start_dm.assert_called_once()
        assert adapter._room_kinds["dm-123"] == RoomKind.DM


# -- 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