hermes-chatto-plugin/plugins/platforms/chatto/test_adapter.py

"""Unit tests for the Chatto platform adapter.

Covers:
  - Protobuf codec (varint, tag, fields, client/server frames, projections)
  - Emoji shortcode conversion
  - Adapter instantiation and properties
  - Registration and requirements
  - Send / reactions / edit / delete (mocked RPC)
  - Typing indicator lifecycle
  - Read state and notifications
  - DM initiation and room creation
  - User lookup (with caching)
  - Presence and custom status
  - Message dispatch (self-echo suppression, handler invocation)
  - Attachment upload (chunked)

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

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

import pytest
import pytest_asyncio

# ── Path setup ────────────────────────────────────────────────────────────
sys.path.insert(0, "/opt/hermes")
sys.path.insert(0, "/root/.hermes/plugins/platforms/chatto")

import adapter as chatto_adapter
from adapter import (
    _encode_varint,
    _decode_varint,
    _encode_tag,
    _encode_field_varint,
    _encode_field_bytes,
    _encode_field_string,
    _encode_submessage,
    _decode_fields,
    _get_first,
    _get_all,
    _encode_client_hello,
    _encode_subscribe_events,
    _encode_ping,
    _encode_client_frame_hello,
    _encode_client_frame_subscribe,
    _encode_client_frame_ping,
    _decode_server_frame,
    _decode_projection_event,
    _decode_projection_operation,
    _decode_room_timeline_event,
    _decode_room_timeline_event_upsert,
    _decode_message_posted,
    _decode_message,
    _decode_thread,
    _decode_timestamp,
    _decode_event_envelope,
    _decode_mention_notification,
    _decode_dm_notification,
    _decode_server_hello,
    _decode_error,
    _EMOJI_TO_SHORTCODE,
    _REALTIME_PROTOCOL_VERSION,
    _MAX_MESSAGE_LENGTH,
    _SEEN_CAP,
    ChattoAdapter,
    check_requirements,
    validate_config,
    register,
)

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import SendResult, MessageEvent, MessageType


# ── 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_URL",
    "CHATTO_LOGIN",
    "CHATTO_PASSWORD",
    "CHATTO_CHANNELS",
    "CHATTO_HOME_CHANNEL",
    "CHATTO_REQUIRE_MENTION",
    "CHATTO_ALLOWED_USERS",
    "CHATTO_ALLOW_ALL_USERS",
]


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 = {"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.  Token is pre-set to avoid
    _ensure_token triggering a real login.  All CHATTO_* env vars are cleared
    first so the config.extra values are not overridden by the environment."""
    _clear_chatto_env()
    cfg = _make_config(**extra_overrides)
    adapter = ChattoAdapter(cfg)
    adapter._token = "test-token"
    adapter._user_id = "bot-user-id"
    adapter._user_login = "hermes_bot"
    adapter._user_display = "Hermes Bot"
    return adapter


# ── Protobuf codec: varint ────────────────────────────────────────────────


class TestVarint:
    """Test _encode_varint / _decode_varint roundtrips."""

    @pytest.mark.parametrize(
        "value",
        [0, 1, 127, 128, 16384, 2**32, 2**63 - 1],
    )
    def test_varint_roundtrip(self, value):
        encoded = _encode_varint(value)
        decoded, offset = _decode_varint(encoded, 0)
        assert decoded == value
        assert offset == len(encoded)

    def test_varint_zero(self):
        assert _encode_varint(0) == b"\x00"

    def test_varint_one(self):
        assert _encode_varint(1) == b"\x01"

    def test_varint_127(self):
        assert _encode_varint(127) == b"\x7f"

    def test_varint_128(self):
        assert _encode_varint(128) == b"\x80\x01"

    def test_varint_300(self):
        # 300 = 0b100101100 → 0xAC 0x02
        assert _encode_varint(300) == b"\xac\x02"

    def test_varint_16384(self):
        # 16384 = 0x4000 → 0x80 0x80 0x01
        assert _encode_varint(16384) == b"\x80\x80\x01"

    def test_decode_varint_truncated(self):
        with pytest.raises(ValueError, match="Truncated"):
            _decode_varint(b"\x80", 0)

    def test_decode_varint_too_long(self):
        # 10 continuation bytes — exceeds 64-bit
        with pytest.raises(ValueError, match="too long"):
            _decode_varint(b"\x80" * 10, 0)


# ── Protobuf codec: tag and field encoders ────────────────────────────────


class TestTagAndFields:
    """Test tag encoding and field-level helpers."""

    def test_encode_tag_field1_varint(self):
        # field 1, wire type 0 → (1<<3)|0 = 8 → 0x08
        assert _encode_tag(1, 0) == b"\x08"

    def test_encode_tag_field2_length_delimited(self):
        # field 2, wire type 2 → (2<<3)|2 = 18 → 0x12
        assert _encode_tag(2, 2) == b"\x12"

    def test_encode_tag_field15_varint(self):
        # field 15, wire type 0 → (15<<3)|0 = 120 → 0x78
        assert _encode_tag(15, 0) == b"\x78"

    def test_encode_field_varint(self):
        result = _encode_field_varint(1, 150)
        # tag(1,0)=0x08 + varint(150)=0x96 0x01
        assert result == b"\x08\x96\x01"

    def test_encode_field_bytes(self):
        result = _encode_field_bytes(2, b"hello")
        # tag(2,2)=0x12 + len(5)=0x05 + "hello"
        assert result == b"\x12\x05hello"

    def test_encode_field_string(self):
        result = _encode_field_string(3, "hi")
        # tag(3,2)=0x1a + len(2)=0x02 + "hi"
        assert result == b"\x1a\x02hi"

    def test_encode_submessage(self):
        inner = _encode_field_varint(1, 42)
        result = _encode_submessage(5, inner)
        # The submessage is length-delimited
        fields = _decode_fields(result)
        assert 5 in fields
        assert isinstance(fields[5][0], bytes)

    def test_decode_fields_varint(self):
        data = _encode_field_varint(1, 42)
        fields = _decode_fields(data)
        assert fields[1] == [42]

    def test_decode_fields_bytes(self):
        data = _encode_field_bytes(2, b"test")
        fields = _decode_fields(data)
        assert fields[2] == [b"test"]

    def test_decode_fields_string(self):
        data = _encode_field_string(3, "hello")
        fields = _decode_fields(data)
        assert fields[3] == [b"hello"]

    def test_decode_fields_multiple(self):
        data = _encode_field_varint(1, 10) + _encode_field_string(2, "abc")
        fields = _decode_fields(data)
        assert fields[1] == [10]
        assert fields[2] == [b"abc"]

    def test_decode_fields_repeated(self):
        data = _encode_field_string(2, "a") + _encode_field_string(2, "b")
        fields = _decode_fields(data)
        assert fields[2] == [b"a", b"b"]

    def test_decode_fields_empty(self):
        fields = _decode_fields(b"")
        assert fields == {}

    def test_get_first(self):
        fields = {1: [10, 20], 2: [b"x"]}
        assert _get_first(fields, 1) == 10
        assert _get_first(fields, 2) == b"x"
        assert _get_first(fields, 99, "default") == "default"

    def test_get_all(self):
        fields = {1: [10, 20]}
        assert _get_all(fields, 1) == [10, 20]
        assert _get_all(fields, 99) == []


# ── Protobuf codec: client hello / subscribe / ping ──────────────────────


class TestClientFrames:
    """Test client-side protobuf message encoders."""

    def test_encode_client_hello_with_token(self):
        msg = _encode_client_hello("my-bearer-token")
        fields = _decode_fields(msg)
        # field 1 = protocol_version (varint)
        assert _get_first(fields, 1) == _REALTIME_PROTOCOL_VERSION
        # field 2 = bearer_token (bytes)
        token_val = _get_first(fields, 2)
        assert isinstance(token_val, bytes)
        assert token_val.decode("utf-8") == "my-bearer-token"

    def test_encode_client_hello_without_token(self):
        msg = _encode_client_hello("")
        fields = _decode_fields(msg)
        assert _get_first(fields, 1) == _REALTIME_PROTOCOL_VERSION
        assert 2 not in fields  # no bearer_token field

    def test_encode_client_hello_protocol_version_is_1(self):
        msg = _encode_client_hello("x")
        fields = _decode_fields(msg)
        assert _get_first(fields, 1) == 1

    def test_encode_subscribe_events_with_rooms(self):
        msg = _encode_subscribe_events(retained_room_ids=["room1", "room2"])
        fields = _decode_fields(msg)
        # field 2 = repeated string
        room_vals = _get_all(fields, 2)
        assert len(room_vals) == 2
        assert room_vals[0].decode("utf-8") == "room1"
        assert room_vals[1].decode("utf-8") == "room2"

    def test_encode_subscribe_events_with_cursor(self):
        msg = _encode_subscribe_events(resume_cursor="cursor123")
        fields = _decode_fields(msg)
        cursor = _get_first(fields, 1)
        assert isinstance(cursor, bytes)
        assert cursor.decode("utf-8") == "cursor123"

    def test_encode_subscribe_events_empty(self):
        msg = _encode_subscribe_events()
        assert msg == b""

    def test_encode_ping_is_empty(self):
        assert _encode_ping() == b""

    def test_encode_client_frame_hello(self):
        hello = _encode_client_hello("token")
        frame = _encode_client_frame_hello(hello)
        fields = _decode_fields(frame)
        # field 1 = hello submessage
        assert 1 in fields
        inner = _get_first(fields, 1)
        assert isinstance(inner, bytes)
        # Decode inner to verify
        inner_fields = _decode_fields(inner)
        assert _get_first(inner_fields, 1) == _REALTIME_PROTOCOL_VERSION

    def test_encode_client_frame_subscribe(self):
        sub = _encode_subscribe_events(retained_room_ids=["r1"])
        frame = _encode_client_frame_subscribe(sub)
        fields = _decode_fields(frame)
        # field 2 = subscribe_events submessage
        assert 2 in fields

    def test_encode_client_frame_ping(self):
        ping = _encode_ping()
        frame = _encode_client_frame_ping(ping)
        fields = _decode_fields(frame)
        # field 3 = ping submessage (empty)
        assert 3 in fields

    def test_client_hello_roundtrip(self):
        """Encode a client hello, wrap in a frame, decode the frame, decode
        the inner hello, and verify values match."""
        original = _encode_client_hello("roundtrip-token")
        frame = _encode_client_frame_hello(original)
        decoded_frame = _decode_server_frame  # not for client frames, but
        # We decode the frame manually
        frame_fields = _decode_fields(frame)
        inner_bytes = _get_first(frame_fields, 1)
        inner_fields = _decode_fields(inner_bytes)
        assert _get_first(inner_fields, 1) == _REALTIME_PROTOCOL_VERSION
        token_raw = _get_first(inner_fields, 2)
        assert token_raw.decode("utf-8") == "roundtrip-token"


# ── Protobuf codec: server frame decoding ────────────────────────────────


class TestServerFrameDecoding:
    """Test _decode_server_frame with synthetic frames."""

    def test_decode_server_hello_frame(self):
        # Build a RealtimeServerHello { protocol_version = 1 }
        hello_inner = _encode_field_varint(1, 1)
        # Wrap in RealtimeServerFrame { hello = 1 }
        frame = _encode_submessage(1, hello_inner)
        result = _decode_server_frame(frame)
        assert result["type"] == "hello"
        assert isinstance(result["data"], bytes)

        # Decode the hello data
        hello = _decode_server_hello(result["data"])
        assert hello["protocolVersion"] == 1

    def test_decode_subscribed_frame(self):
        # Build an empty submessage for field 2 (subscribed)
        frame = _encode_submessage(2, b"")
        result = _decode_server_frame(frame)
        assert result["type"] == "subscribed"

    def test_decode_error_frame(self):
        # Build RealtimeError { message = "bad", code = 500 }
        error_inner = _encode_field_string(1, "bad") + _encode_field_varint(2, 500)
        # Wrap in RealtimeServerFrame { error = 5 }
        frame = _encode_submessage(5, error_inner)
        result = _decode_server_frame(frame)
        assert result["type"] == "error"
        decoded = _decode_error(result["data"])
        assert decoded["message"] == "bad"
        assert decoded["code"] == 500

    def test_decode_pong_frame(self):
        frame = _encode_submessage(7, b"")
        result = _decode_server_frame(frame)
        assert result["type"] == "pong"

    def test_decode_heartbeat_frame(self):
        frame = _encode_submessage(4, b"")
        result = _decode_server_frame(frame)
        assert result["type"] == "heartbeat"

    def test_decode_close_frame(self):
        close_inner = _encode_field_string(1, "bye")
        frame = _encode_submessage(6, close_inner)
        result = _decode_server_frame(frame)
        assert result["type"] == "close"

    def test_decode_caught_up_frame(self):
        frame = _encode_submessage(8, b"")
        result = _decode_server_frame(frame)
        assert result["type"] == "caught_up"

    def test_decode_unknown_frame(self):
        # Empty frame → no fields → unknown
        result = _decode_server_frame(b"")
        assert result["type"] == "unknown"
        assert result["data"] is None


# ── Protobuf codec: projection event decoding ────────────────────────────


class TestProjectionEventDecoding:
    """Test _decode_projection_event and related decoders."""

    def test_decode_projection_event_basic(self):
        # Build RealtimeProjectionEvent {
        #   id = "evt1",
        #   actor_id = "user1",
        #   resume_cursor = "cursor1"
        # }
        proj = (
            _encode_field_string(1, "evt1")
            + _encode_field_string(3, "user1")
            + _encode_field_string(4, "cursor1")
        )
        result = _decode_projection_event(proj)
        assert result["id"] == "evt1"
        assert result["actor_id"] == "user1"
        assert result["resume_cursor"] == "cursor1"
        assert result["operations"] == []

    def test_decode_projection_event_with_timestamp(self):
        ts_inner = _encode_field_varint(1, 1700000000)  # seconds
        proj = (
            _encode_field_string(1, "evt2")
            + _encode_submessage(2, ts_inner)
        )
        result = _decode_projection_event(proj)
        assert result["id"] == "evt2"
        assert "1700000000" not in result["created_at"]  # should be ISO format
        assert "T" in result["created_at"]  # ISO format has T separator

    def test_decode_projection_event_with_operation(self):
        # Build a room_timeline_event_upsert operation
        room_id = _encode_field_string(1, "room123")
        # Build a minimal RoomTimelineEvent
        event_inner = _encode_field_string(1, "evt456")
        event_upsert = room_id + _encode_submessage(2, event_inner)
        # Wrap in RealtimeProjectionOperation { room_timeline_event_upsert = 10 }
        op = _encode_submessage(10, event_upsert)
        # Wrap in RealtimeProjectionEvent { operations = 5 }
        proj = _encode_field_string(1, "proj1") + _encode_submessage(5, op)

        result = _decode_projection_event(proj)
        assert len(result["operations"]) == 1
        op_result = result["operations"][0]
        assert op_result["type"] == "room_timeline_event_upsert"
        assert op_result["room_id"] == "room123"
        assert op_result["event"]["id"] == "evt456"

    def test_decode_projection_operation_unknown(self):
        # An operation with field 1 (room_upsert) — not handled specifically
        op = _encode_submessage(1, _encode_field_string(1, "room1"))
        result = _decode_projection_operation(op)
        assert result["type"] == "room_upsert"

    def test_decode_projection_operation_empty(self):
        result = _decode_projection_operation(b"")
        assert result["type"] == "empty"


# ── Protobuf codec: room timeline event decoding ─────────────────────────


class TestRoomTimelineEventDecoding:
    """Test _decode_room_timeline_event and _decode_message."""

    def test_decode_room_timeline_event_basic(self):
        # Build RoomTimelineEvent {
        #   id = "evt1", room_id = "room1", kind = 1 (message_posted)
        # }
        ts_inner = _encode_field_varint(1, 1700000000)
        event = (
            _encode_field_string(1, "evt1")
            + _encode_submessage(2, ts_inner)
            + _encode_field_string(3, "room1")
            + _encode_field_varint(4, 1)
        )
        result = _decode_room_timeline_event(event)
        assert result["id"] == "evt1"
        assert result["roomId"] == "room1"
        assert result["kind"] == 1
        assert "T" in result["createdAt"]

    def test_decode_room_timeline_event_with_message(self):
        # Build a Message { id="m1", room_id="r1", actor_id="u1", body="hello" }
        ts_inner = _encode_field_varint(1, 1700000000)
        msg = (
            _encode_field_string(1, "m1")
            + _encode_field_string(2, "r1")
            + _encode_field_string(3, "u1")
            + _encode_field_string(4, "hello")
            + _encode_submessage(5, ts_inner)
        )
        # Wrap in MessagePosted { message = 1 }
        posted = _encode_submessage(1, msg)
        # Wrap in RoomTimelineEvent { message_posted = 5 }
        event = (
            _encode_field_string(1, "evt1")
            + _encode_field_string(3, "r1")
            + _encode_submessage(5, posted)
        )
        result = _decode_room_timeline_event(event)
        assert result["id"] == "evt1"
        assert result["messagePosted"] is not None
        assert result["messagePosted"]["message"]["id"] == "m1"
        assert result["messagePosted"]["message"]["body"] == "hello"
        assert result["messagePosted"]["message"]["actorId"] == "u1"

    def test_decode_message_with_thread(self):
        # Build Message with a thread
        thread_inner = _encode_field_string(1, "thread-root-123")
        msg = (
            _encode_field_string(1, "m1")
            + _encode_field_string(2, "r1")
            + _encode_field_string(3, "u1")
            + _encode_field_string(4, "threaded reply")
            + _encode_submessage(8, thread_inner)
        )
        result = _decode_message(msg)
        assert result["id"] == "m1"
        assert result["body"] == "threaded reply"
        assert result["thread"]["threadRootEventId"] == "thread-root-123"

    def test_decode_message_with_login_and_display(self):
        msg = (
            _encode_field_string(1, "m1")
            + _encode_field_string(2, "r1")
            + _encode_field_string(3, "u1")
            + _encode_field_string(4, "hi")
            + _encode_field_string(6, "alice")
            + _encode_field_string(7, "Alice Smith")
        )
        result = _decode_message(msg)
        assert result["actorLogin"] == "alice"
        assert result["actorDisplayName"] == "Alice Smith"

    def test_decode_thread(self):
        thread = _encode_field_string(1, "thread-abc")
        result = _decode_thread(thread)
        assert result["threadRootEventId"] == "thread-abc"

    def test_decode_thread_empty(self):
        result = _decode_thread(b"")
        assert result["threadRootEventId"] == ""

    def test_decode_message_posted_empty(self):
        result = _decode_message_posted(b"")
        assert result == {"message": {}}


# ── Protobuf codec: timestamp decoding ────────────────────────────────────


class TestTimestampDecoding:
    """Test _decode_timestamp."""

    def test_decode_timestamp_basic(self):
        ts = _encode_field_varint(1, 1700000000)
        result = _decode_timestamp(ts)
        assert "2023" in result  # Nov 14, 2023
        assert result.endswith("Z")

    def test_decode_timestamp_with_nanos(self):
        ts = _encode_field_varint(1, 1700000000) + _encode_field_varint(2, 500000)
        result = _decode_timestamp(ts)
        assert "2023" in result
        # 500000 nanos = 0.5 seconds → microsecond=500000 → ".000500" in ISO
        assert ".000500" in result

    def test_decode_timestamp_zero(self):
        ts = b""
        result = _decode_timestamp(ts)
        assert result == ""

    def test_decode_timestamp_only_nanos(self):
        # No seconds, just nanos — should return "" since seconds == 0
        ts = _encode_field_varint(2, 1000000)
        result = _decode_timestamp(ts)
        assert result == ""


# ── Protobuf codec: event envelope (transient events) ────────────────────


class TestEventEnvelopeDecoding:
    """Test _decode_event_envelope for mention and DM notifications."""

    def test_decode_mention_notification(self):
        # Build MentionNotification { room_id = "r1", event_id = "e1" }
        mention_inner = _encode_field_string(1, "r1") + _encode_field_string(2, "e1")
        # Wrap in RealtimeEventEnvelope { mention_notification = 88 }
        envelope = (
            _encode_field_string(1, "envelope1")
            + _encode_submessage(88, mention_inner)
        )
        result = _decode_event_envelope(envelope)
        assert result["id"] == "envelope1"
        assert result["type"] == "mention_notification"
        assert result["data"]["roomId"] == "r1"
        assert result["data"]["eventId"] == "e1"

    def test_decode_dm_notification(self):
        # Build NewDirectMessageNotification { room_id = "r2", event_id = "e2" }
        dm_inner = _encode_field_string(1, "r2") + _encode_field_string(2, "e2")
        # Wrap in RealtimeEventEnvelope { new_direct_message_notification = 89 }
        envelope = (
            _encode_field_string(1, "envelope2")
            + _encode_submessage(89, dm_inner)
        )
        result = _decode_event_envelope(envelope)
        assert result["id"] == "envelope2"
        assert result["type"] == "new_direct_message_notification"
        assert result["data"]["roomId"] == "r2"
        assert result["data"]["eventId"] == "e2"

    def test_decode_event_envelope_unknown(self):
        envelope = _encode_field_string(1, "env3")
        result = _decode_event_envelope(envelope)
        assert result["id"] == "env3"
        assert result["type"] == "unknown"
        assert result["data"] == {}

    def test_decode_mention_notification_directly(self):
        mention_inner = _encode_field_string(1, "roomX") + _encode_field_string(2, "evtX")
        result = _decode_mention_notification(mention_inner)
        assert result["roomId"] == "roomX"
        assert result["eventId"] == "evtX"

    def test_decode_dm_notification_directly(self):
        dm_inner = _encode_field_string(1, "roomY") + _encode_field_string(2, "evtY")
        result = _decode_dm_notification(dm_inner)
        assert result["roomId"] == "roomY"
        assert result["eventId"] == "evtY"


# ── Emoji shortcode conversion ───────────────────────────────────────────


class TestEmojiShortcode:
    """Test _emoji_to_shortcode static method and the emoji mapping."""

    @pytest.mark.parametrize(
        "emoji,shortcode",
        [
            ("👍", "thumbsup"),
            ("👎", "thumbsdown"),
            ("❤️", "heart"),
            ("❤", "heart"),
            ("✅", "white_check_mark"),
            ("❌", "x"),
            ("👀", "eyes"),
            ("🎉", "tada"),
            ("😂", "joy"),
            ("🚀", "rocket"),
            ("🔥", "fire"),
            ("💯", "100"),
            ("🤔", "thinking"),
            ("👏", "clap"),
            ("🙏", "pray"),
            ("😅", "sweat_smile"),
            ("😴", "sleeping"),
            ("⏳", "hourglass"),
        ],
    )
    def test_known_emoji_to_shortcode(self, emoji, shortcode):
        assert ChattoAdapter._emoji_to_shortcode(emoji) == shortcode

    def test_unknown_emoji_passes_through(self):
        # 🦀 (crab) is not in the mapping
        assert ChattoAdapter._emoji_to_shortcode("🦀") == "🦀"

    def test_shortcode_passes_through(self):
        assert ChattoAdapter._emoji_to_shortcode("thumbsup") == "thumbsup"

    def test_empty_string_passes_through(self):
        assert ChattoAdapter._emoji_to_shortcode("") == ""

    def test_emoji_mapping_completeness(self):
        """Verify the mapping dict has the expected keys."""
        assert "👍" in _EMOJI_TO_SHORTCODE
        assert _EMOJI_TO_SHORTCODE["👍"] == "thumbsup"
        assert _EMOJI_TO_SHORTCODE["❤️"] == "heart"


# ── Adapter instantiation and properties ──────────────────────────────────


class TestAdapterInstantiation:
    """Test ChattoAdapter creation and property values."""

    def test_platform_name(self):
        adapter = _make_adapter()
        assert adapter.platform_name == "chatto"

    def test_supports_markdown(self):
        adapter = _make_adapter()
        assert adapter.supports_markdown is True

    def test_supports_reactions(self):
        adapter = _make_adapter()
        assert adapter.supports_reactions is True

    def test_supports_threads(self):
        adapter = _make_adapter()
        assert adapter.supports_threads is True

    def test_max_message_length(self):
        adapter = _make_adapter()
        assert adapter.MAX_MESSAGE_LENGTH == 10000

    def test_splits_long_messages(self):
        adapter = _make_adapter()
        assert adapter.splits_long_messages is True

    def test_typing_tasks_empty(self):
        adapter = _make_adapter()
        assert adapter._typing_tasks == {}

    def test_user_cache_empty(self):
        adapter = _make_adapter()
        assert adapter._user_cache == {}

    def test_base_url_from_extra(self):
        adapter = _make_adapter()
        assert adapter._base_url == "https://chat.example.com"

    def test_channels_from_extra(self):
        adapter = _make_adapter()
        assert adapter._channel_ids == ["room1"]

    def test_require_mention_default(self):
        adapter = _make_adapter(require_mention=False)
        assert adapter._require_mention is False

    def test_is_base_platform_adapter(self):
        from gateway.platforms.base import BasePlatformAdapter

        adapter = _make_adapter()
        assert isinstance(adapter, BasePlatformAdapter)


# ── Registration and requirements ─────────────────────────────────────────


class TestRegistration:
    """Test plugin registration and requirements checking."""

    def test_register_calls_register_platform(self):
        ctx = _MockPluginContext()
        register(ctx)
        assert "chatto" in ctx.registered_names
        kwargs = ctx.registered_kwargs
        assert kwargs["name"] == "chatto"
        assert kwargs["label"] == "Chatto"
        assert callable(kwargs["adapter_factory"])
        assert callable(kwargs["check_fn"])

    def test_register_adapter_factory_creates_adapter(self):
        ctx = _MockPluginContext()
        register(ctx)
        kwargs = ctx.registered_kwargs
        cfg = _make_config()
        adapter = kwargs["adapter_factory"](cfg)
        assert adapter is not None
        assert isinstance(adapter, ChattoAdapter)

    def test_check_requirements_true_when_env_set(self, monkeypatch):
        monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
        monkeypatch.setenv("CHATTO_LOGIN", "user")
        monkeypatch.setenv("CHATTO_PASSWORD", "pass")
        assert check_requirements() is True

    def test_check_requirements_false_when_url_missing(self, monkeypatch):
        monkeypatch.delenv("CHATTO_URL", raising=False)
        monkeypatch.setenv("CHATTO_LOGIN", "user")
        monkeypatch.setenv("CHATTO_PASSWORD", "pass")
        assert check_requirements() is False

    def test_check_requirements_false_when_login_missing(self, monkeypatch):
        monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
        monkeypatch.delenv("CHATTO_LOGIN", raising=False)
        monkeypatch.setenv("CHATTO_PASSWORD", "pass")
        assert check_requirements() is False

    def test_check_requirements_false_when_password_missing(self, monkeypatch):
        monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
        monkeypatch.setenv("CHATTO_LOGIN", "user")
        monkeypatch.delenv("CHATTO_PASSWORD", raising=False)
        assert check_requirements() is False

    def test_check_requirements_false_all_missing(self, monkeypatch):
        monkeypatch.delenv("CHATTO_URL", raising=False)
        monkeypatch.delenv("CHATTO_LOGIN", raising=False)
        monkeypatch.delenv("CHATTO_PASSWORD", raising=False)
        assert check_requirements() is False

    def test_validate_config_with_extra_url(self):
        cfg = MagicMock()
        cfg.extra = {"url": "https://chat.example.com"}
        # Also need login/password from env
        with patch.dict(os.environ, {"CHATTO_LOGIN": "u", "CHATTO_PASSWORD": "p"}):
            assert validate_config(cfg) is True

    def test_validate_config_missing_url(self):
        cfg = MagicMock()
        cfg.extra = {}
        with patch.dict(os.environ, {"CHATTO_LOGIN": "u", "CHATTO_PASSWORD": "p"}, clear=False):
            # Remove CHATTO_URL if set
            os.environ.pop("CHATTO_URL", None)
            assert validate_config(cfg) is False


# ── Send method (mocked RPC) ──────────────────────────────────────────────


class TestSend:
    """Test the send() method with mocked _rpc."""

    @pytest.mark.asyncio
    async def test_send_basic(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt123"}}))
        result = await adapter.send("room1", "Hello world")
        assert result.success is True
        assert result.message_id == "evt123"

    @pytest.mark.asyncio
    async def test_send_with_reply_to(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt456"}}))
        adapter._follow_thread = AsyncMock()
        await adapter.send("room1", "Reply", reply_to="thread-root-1")
        # Verify RPC body contains threadRootEventId
        call_args = adapter._rpc.call_args
        body = call_args.kwargs.get("body") or call_args.args[1]
        assert body["threadRootEventId"] == "thread-root-1"

    @pytest.mark.asyncio
    async def test_send_with_metadata_thread_id(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt789"}}))
        adapter._follow_thread = AsyncMock()
        await adapter.send("room1", "Reply", metadata={"thread_id": "thread456"})
        call_args = adapter._rpc.call_args
        body = call_args.kwargs.get("body") or call_args.args[1]
        assert body["threadRootEventId"] == "thread456"

    @pytest.mark.asyncio
    async def test_send_empty_content(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock()
        result = await adapter.send("room1", "")
        assert result.success is False
        assert "Empty" in (result.error or "")
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_send_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {"error": "server error"}))
        result = await adapter.send("room1", "Hello")
        assert result.success is False
        assert "server error" in (result.error or "")

    @pytest.mark.asyncio
    async def test_send_long_message_split(self):
        adapter = _make_adapter()
        # Return different message IDs for each call
        call_count = [0]

        async def mock_rpc(path, body, **kwargs):
            call_count[0] += 1
            return 200, {"message": {"id": f"evt-{call_count[0]}"}}

        adapter._rpc = AsyncMock(side_effect=mock_rpc)
        adapter._follow_thread = AsyncMock()
        # Create content longer than MAX_MESSAGE_LENGTH (10000)
        long_content = "A" * 12000
        result = await adapter.send("room1", long_content)
        assert result.success is True
        assert result.message_id == "evt-1"
        # Should have been called multiple times
        assert adapter._rpc.call_count > 1

    @pytest.mark.asyncio
    async def test_send_marks_seen(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "new_evt"}}))
        await adapter.send("room1", "Hello")
        assert "new_evt" in adapter._seen.get("room1", {})


# ── Reactions (mocked RPC) ────────────────────────────────────────────────


class TestReactions:
    """Test send_reaction and remove_reaction."""

    @pytest.mark.asyncio
    async def test_send_reaction_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"added": True}))
        result = await adapter.send_reaction("room1", "evt1", "👍")
        assert result is True
        # Verify emoji was converted to shortcode
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["emoji"] == "thumbsup"

    @pytest.mark.asyncio
    async def test_send_reaction_heart(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        await adapter.send_reaction("room1", "evt1", "❤️")
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["emoji"] == "heart"

    @pytest.mark.asyncio
    async def test_remove_reaction_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.remove_reaction("room1", "evt1", "❤️")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["emoji"] == "heart"

    @pytest.mark.asyncio
    async def test_send_reaction_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.send_reaction("room1", "evt1", "👍")
        assert result is False

    @pytest.mark.asyncio
    async def test_remove_reaction_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.remove_reaction("room1", "evt1", "❤️")
        assert result is False

    @pytest.mark.asyncio
    async def test_send_reaction_exception(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(side_effect=Exception("network error"))
        result = await adapter.send_reaction("room1", "evt1", "👍")
        assert result is False


# ── Message edit/delete (mocked RPC) ──────────────────────────────────────


class TestEditDelete:
    """Test edit_message and delete_message."""

    @pytest.mark.asyncio
    async def test_edit_message_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.edit_message("room1", "evt1", "new text")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["body"] == "new text"
        assert body["eventId"] == "evt1"

    @pytest.mark.asyncio
    async def test_edit_message_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.edit_message("room1", "evt1", "new text")
        assert result is False

    @pytest.mark.asyncio
    async def test_delete_message_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.delete_message("room1", "evt1")
        assert result is True

    @pytest.mark.asyncio
    async def test_delete_message_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.delete_message("room1", "evt1")
        assert result is False

    @pytest.mark.asyncio
    async def test_edit_message_exception(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(side_effect=Exception("boom"))
        result = await adapter.edit_message("room1", "evt1", "text")
        assert result is False

    @pytest.mark.asyncio
    async def test_delete_message_exception(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(side_effect=Exception("boom"))
        result = await adapter.delete_message("room1", "evt1")
        assert result is False


# ── Typing indicator lifecycle ─────────────────────────────────────────────


class TestTypingIndicator:
    """Test send_typing and stop_typing lifecycle."""

    @pytest.mark.asyncio
    async def test_send_typing_creates_task(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        await adapter.send_typing("room1")
        assert "room1" in adapter._typing_tasks
        assert isinstance(adapter._typing_tasks["room1"], asyncio.Task)
        # Clean up
        await adapter.stop_typing("room1")

    @pytest.mark.asyncio
    async def test_send_typing_no_duplicate(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        await adapter.send_typing("room1")
        first_task = adapter._typing_tasks["room1"]
        await adapter.send_typing("room1")
        assert adapter._typing_tasks["room1"] is first_task
        await adapter.stop_typing("room1")

    @pytest.mark.asyncio
    async def test_stop_typing_cancels_task(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        await adapter.send_typing("room1")
        assert "room1" in adapter._typing_tasks
        await adapter.stop_typing("room1")
        assert "room1" not in adapter._typing_tasks

    @pytest.mark.asyncio
    async def test_stop_typing_when_not_running(self):
        adapter = _make_adapter()
        # Should not raise even if no task exists
        await adapter.stop_typing("room1")

    @pytest.mark.asyncio
    async def test_typing_loop_calls_rpc(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        await adapter.send_typing("room1")
        # Allow the loop to run one iteration
        await asyncio.sleep(0.05)
        # The typing loop should have called _rpc at least once
        assert adapter._rpc.call_count >= 1
        # Verify it was called with the typing path
        first_call = adapter._rpc.call_args_list[0]
        path = first_call.args[0] if len(first_call.args) > 0 else first_call.kwargs.get("path")
        assert "UpdateTypingIndicator" in path
        body = first_call.args[1] if len(first_call.args) > 1 else first_call.kwargs.get("body")
        assert body["roomId"] == "room1"
        assert body["typing"] is True
        await adapter.stop_typing("room1")


# ── Read state and notifications (mocked RPC) ─────────────────────────────


class TestReadStateAndNotifications:
    """Test mark_room_as_read, mark_thread_as_read, dismiss notifications."""

    @pytest.mark.asyncio
    async def test_mark_room_as_read_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.mark_room_as_read("room1")
        assert result is True

    @pytest.mark.asyncio
    async def test_mark_room_as_read_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.mark_room_as_read("room1")
        assert result is False

    @pytest.mark.asyncio
    async def test_mark_thread_as_read_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.mark_thread_as_read("room1", "thread-root-1")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["roomId"] == "room1"
        assert body["threadRootEventId"] == "thread-root-1"

    @pytest.mark.asyncio
    async def test_mark_thread_as_read_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.mark_thread_as_read("room1", "thread-root-1")
        assert result is False

    @pytest.mark.asyncio
    async def test_dismiss_all_notifications_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.dismiss_all_notifications()
        assert result is True

    @pytest.mark.asyncio
    async def test_dismiss_all_notifications_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.dismiss_all_notifications()
        assert result is False

    @pytest.mark.asyncio
    async def test_dismiss_notification_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.dismiss_notification("notif123")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["notificationId"] == "notif123"

    @pytest.mark.asyncio
    async def test_dismiss_notification_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.dismiss_notification("notif123")
        assert result is False


# ── DM initiation and room creation (mocked RPC) ──────────────────────────


class TestDMAndRoomCreation:
    """Test start_dm and create_room."""

    @pytest.mark.asyncio
    async def test_start_dm_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "room789"}}))
        result = await adapter.start_dm("user123")
        assert result == "room789"
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["participantIds"] == ["user123"]

    @pytest.mark.asyncio
    async def test_start_dm_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.start_dm("user123")
        assert result is None

    @pytest.mark.asyncio
    async def test_start_dm_no_room_id(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"room": {}}))
        result = await adapter.start_dm("user123")
        assert result is None

    @pytest.mark.asyncio
    async def test_start_dm_sets_room_kind(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "dm-room-1"}}))
        await adapter.start_dm("user123")
        assert adapter._room_kinds["dm-room-1"] == "ROOM_KIND_DM"

    @pytest.mark.asyncio
    async def test_create_room_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "room789"}}))
        result = await adapter.create_room("test-room", "description")
        assert result == "room789"
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["name"] == "test-room"
        assert body["description"] == "description"

    @pytest.mark.asyncio
    async def test_create_room_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.create_room("test-room", "description")
        assert result is None

    @pytest.mark.asyncio
    async def test_create_room_sets_room_kind(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "group-room-1"}}))
        await adapter.create_room("test-room")
        assert adapter._room_kinds["group-room-1"] == "ROOM_KIND_GROUP"
        assert adapter._room_names["group-room-1"] == "test-room"


# ── User lookup (mocked RPC) ──────────────────────────────────────────────


class TestUserLookup:
    """Test get_user, list_users, batch_get_users with caching."""

    @pytest.mark.asyncio
    async def test_get_user_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(
            return_value=(200, {"user": {"id": "u1", "login": "alice", "displayName": "Alice"}})
        )
        result = await adapter.get_user("u1")
        assert result is not None
        assert result["id"] == "u1"
        assert result["login"] == "alice"

    @pytest.mark.asyncio
    async def test_get_user_caches(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(
            return_value=(200, {"user": {"id": "u1", "login": "alice"}})
        )
        await adapter.get_user("u1")
        # Second call should use cache — no second RPC
        await adapter.get_user("u1")
        assert adapter._rpc.call_count == 1
        assert "u1" in adapter._user_cache

    @pytest.mark.asyncio
    async def test_get_user_empty_id(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock()
        result = await adapter.get_user("")
        assert result is None
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_get_user_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(404, {}))
        result = await adapter.get_user("u1")
        assert result is None

    @pytest.mark.asyncio
    async def test_get_user_exception(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(side_effect=Exception("boom"))
        result = await adapter.get_user("u1")
        assert result is None

    @pytest.mark.asyncio
    async def test_list_users_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(
            return_value=(200, {"users": [{"id": "u1"}, {"id": "u2"}]})
        )
        result = await adapter.list_users()
        assert len(result) == 2
        assert result[0]["id"] == "u1"
        # Should cache all returned users
        assert "u1" in adapter._user_cache
        assert "u2" in adapter._user_cache

    @pytest.mark.asyncio
    async def test_list_users_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.list_users()
        assert result == []

    @pytest.mark.asyncio
    async def test_batch_get_users_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(
            return_value=(200, {"users": [{"id": "u1"}, {"id": "u2"}]})
        )
        result = await adapter.batch_get_users(["u1", "u2"])
        assert len(result) == 2
        # Should cache results
        assert "u1" in adapter._user_cache
        assert "u2" in adapter._user_cache

    @pytest.mark.asyncio
    async def test_batch_get_users_empty_list(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock()
        result = await adapter.batch_get_users([])
        assert result == []
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_batch_get_users_uses_cache(self):
        adapter = _make_adapter()
        # Pre-populate cache
        adapter._user_cache["u1"] = {"id": "u1", "login": "alice"}
        adapter._rpc = AsyncMock(
            return_value=(200, {"users": [{"id": "u2"}]})
        )
        result = await adapter.batch_get_users(["u1", "u2"])
        assert len(result) == 2
        # Should have only fetched u2 from server
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["userIds"] == ["u2"]

    @pytest.mark.asyncio
    async def test_batch_get_users_all_cached(self):
        adapter = _make_adapter()
        adapter._user_cache["u1"] = {"id": "u1"}
        adapter._rpc = AsyncMock()
        result = await adapter.batch_get_users(["u1"])
        assert len(result) == 1
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_batch_get_users_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.batch_get_users(["u1"])
        assert result == []


# ── Presence and custom status (mocked RPC) ──────────────────────────────


class TestPresenceAndStatus:
    """Test set_presence, set_custom_status, clear_custom_status."""

    @pytest.mark.asyncio
    async def test_set_presence_online(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_presence("online")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["status"] == 1

    @pytest.mark.asyncio
    async def test_set_presence_dnd(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_presence("dnd")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["status"] == 3

    @pytest.mark.asyncio
    async def test_set_presence_away(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_presence("away")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["status"] == 2

    @pytest.mark.asyncio
    async def test_set_presence_do_not_disturb(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_presence("do_not_disturb")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["status"] == 3

    @pytest.mark.asyncio
    async def test_set_presence_unknown(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_presence("invisible")
        assert result is False
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_set_presence_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.set_presence("online")
        assert result is False

    @pytest.mark.asyncio
    async def test_set_custom_status_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.set_custom_status("Processing...")
        assert result is True
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert body["status"] == "Processing..."

    @pytest.mark.asyncio
    async def test_set_custom_status_empty(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock()
        result = await adapter.set_custom_status("")
        assert result is False
        adapter._rpc.assert_not_called()

    @pytest.mark.asyncio
    async def test_set_custom_status_truncates(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        long_status = "A" * 200
        await adapter.set_custom_status(long_status)
        call_args = adapter._rpc.call_args
        body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
        assert len(body["status"]) == 100

    @pytest.mark.asyncio
    async def test_set_custom_status_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.set_custom_status("test")
        assert result is False

    @pytest.mark.asyncio
    async def test_clear_custom_status_success(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        result = await adapter.clear_custom_status()
        assert result is True

    @pytest.mark.asyncio
    async def test_clear_custom_status_failure(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(500, {}))
        result = await adapter.clear_custom_status()
        assert result is False


# ── Message dispatch (mocked) ─────────────────────────────────────────────


class TestMessageDispatch:
    """Test _dispatch_message with mocked handler.

    The base class ``handle_message`` does complex session management and
    spawns background tasks, so we mock it to verify _dispatch_message calls
    it with the right MessageEvent.
    """

    @pytest.mark.asyncio
    async def test_dispatch_message_calls_handler(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_names["room1"] = "General"
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = False  # don't require mention for this test

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello bot",
            "createdAt": "",
            "actorLogin": "alice",
            "actorDisplayName": "Alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        assert adapter.handle_message.called
        event = adapter.handle_message.call_args.args[0]
        assert event.text == "Hello bot"
        assert event.message_id == "evt1"

    @pytest.mark.asyncio
    async def test_dispatch_message_self_echo_suppressed(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "bot-user-id",  # matches adapter._user_id
            "body": "My own message",
            "createdAt": "",
            "actorLogin": "hermes_bot",
            "actorDisplayName": "Hermes Bot",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        adapter.handle_message.assert_not_called()

    @pytest.mark.asyncio
    async def test_dispatch_message_empty_body_skipped(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "",
            "createdAt": "",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        adapter.handle_message.assert_not_called()

    @pytest.mark.asyncio
    async def test_dispatch_message_no_handler(self):
        adapter = _make_adapter()
        adapter._message_handler = None
        adapter._rpc = AsyncMock(return_value=(200, {}))
        # Should not raise
        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello",
            "createdAt": "",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")

    @pytest.mark.asyncio
    async def test_dispatch_message_mark_room_read_called(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = False  # don't require mention for this test

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        # mark_room_as_read and dismiss_all_notifications should have been called
        # via _rpc — check for MarkRoomAsRead and DismissAllNotifications paths
        rpc_paths = [c.args[0] for c in adapter._rpc.call_args_list if len(c.args) > 0]
        assert any("MarkRoomAsRead" in p for p in rpc_paths)
        assert any("DismissAllNotifications" in p for p in rpc_paths)

    @pytest.mark.asyncio
    async def test_dispatch_message_dm_always_responds(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_DM"
        adapter._require_mention = True  # even with require_mention, DMs respond

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello without mention",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        assert adapter.handle_message.called

    @pytest.mark.asyncio
    async def test_dispatch_message_require_mention_no_mention_skipped(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = True

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello without mention",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        adapter.handle_message.assert_not_called()

    @pytest.mark.asyncio
    async def test_dispatch_message_with_mention(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = True

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "@hermes_bot do something",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        assert adapter.handle_message.called

    @pytest.mark.asyncio
    async def test_dispatch_message_with_thread(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = False  # don't require mention for this test

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "Hello",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {"threadRootEventId": "thread-root-123"},
        }
        await adapter._dispatch_message(msg, "room1")
        assert adapter.handle_message.called
        # Verify the event was created with the thread_id
        event = adapter.handle_message.call_args.args[0]
        assert event.source.thread_id == "thread-root-123"

    @pytest.mark.asyncio
    async def test_dispatch_message_strips_mention_prefix(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = True

        msg = {
            "id": "evt1",
            "roomId": "room1",
            "actorId": "user1",
            "body": "@hermes_bot please help",
            "createdAt": "",
            "actorLogin": "alice",
            "thread": {},
        }
        await adapter._dispatch_message(msg, "room1")
        event = adapter.handle_message.call_args.args[0]
        # The mention prefix should be stripped
        assert not event.text.startswith("@hermes_bot")
        assert "please help" in event.text


# ── Processing lifecycle hooks ─────────────────────────────────────────────


class TestProcessingLifecycle:
    """Test on_processing_start and on_processing_complete reaction hooks."""

    @pytest.mark.asyncio
    async def test_on_processing_start_adds_eyes(self):
        adapter = _make_adapter()
        adapter.send_reaction = AsyncMock(return_value=True)
        event = MagicMock()
        event.message_id = "evt1"
        event.source = MagicMock()
        event.source.chat_id = "room1"
        event.raw_message = {}
        await adapter.on_processing_start(event)
        adapter.send_reaction.assert_called_once_with("room1", "evt1", "👀")

    @pytest.mark.asyncio
    async def test_on_processing_complete_success(self):
        from gateway.platforms.base import ProcessingOutcome

        adapter = _make_adapter()
        adapter.send_reaction = AsyncMock(return_value=True)
        adapter.remove_reaction = AsyncMock(return_value=True)
        event = MagicMock()
        event.message_id = "evt1"
        event.source = MagicMock()
        event.source.chat_id = "room1"
        event.raw_message = {}
        await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS)
        adapter.remove_reaction.assert_called_once_with("room1", "evt1", "👀")
        adapter.send_reaction.assert_called_once_with("room1", "evt1", "✅")

    @pytest.mark.asyncio
    async def test_on_processing_complete_failure(self):
        from gateway.platforms.base import ProcessingOutcome

        adapter = _make_adapter()
        adapter.send_reaction = AsyncMock(return_value=True)
        adapter.remove_reaction = AsyncMock(return_value=True)
        event = MagicMock()
        event.message_id = "evt1"
        event.source = MagicMock()
        event.source.chat_id = "room1"
        event.raw_message = {}
        await adapter.on_processing_complete(event, ProcessingOutcome.FAILURE)
        adapter.send_reaction.assert_called_once_with("room1", "evt1", "❌")

    @pytest.mark.asyncio
    async def test_reactions_disabled(self, monkeypatch):
        adapter = _make_adapter()
        monkeypatch.setenv("CHATTO_REACTIONS", "false")
        adapter.send_reaction = AsyncMock()
        event = MagicMock()
        event.message_id = "evt1"
        event.source = MagicMock()
        event.source.chat_id = "room1"
        event.raw_message = {}
        await adapter.on_processing_start(event)
        adapter.send_reaction.assert_not_called()


# ── Attachment upload (mocked RPC) ────────────────────────────────────────


class TestAttachmentUpload:
    """Test the chunked asset upload flow."""

    @pytest.mark.asyncio
    async def test_upload_asset_success(self):
        adapter = _make_adapter()

        # Create a temporary file
        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
            f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
            tmp_path = f.name

        try:
            rpc_responses = [
                (200, {"upload": {"id": "upload-1"}}),  # CreateUpload
                (200, {}),  # UploadChunk
                (200, {"asset": {"id": "asset-1"}}),  # CompleteUpload
            ]
            adapter._rpc = AsyncMock(side_effect=rpc_responses)

            asset_id = await adapter._upload_asset("room1", tmp_path)
            assert asset_id == "asset-1"
            assert adapter._rpc.call_count == 3

            # Verify the first call was CreateUpload
            first_call = adapter._rpc.call_args_list[0]
            path = first_call.args[0]
            assert "CreateUpload" in path
            body = first_call.args[1]
            assert body["roomId"] == "room1"
            assert "sha256" in body

            # Verify last call was CompleteUpload
            last_call = adapter._rpc.call_args_list[-1]
            path = last_call.args[0]
            assert "CompleteUpload" in path
        finally:
            os.unlink(tmp_path)

    @pytest.mark.asyncio
    async def test_upload_asset_create_fails(self):
        adapter = _make_adapter()

        with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as f:
            f.write(b"test data")
            tmp_path = f.name

        try:
            adapter._rpc = AsyncMock(return_value=(500, {"error": "fail"}))
            result = await adapter._upload_asset("room1", tmp_path)
            assert result is None
        finally:
            os.unlink(tmp_path)

    @pytest.mark.asyncio
    async def test_upload_asset_empty_file(self):
        adapter = _make_adapter()

        with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as f:
            f.write(b"")
            tmp_path = f.name

        try:
            adapter._rpc = AsyncMock()
            result = await adapter._upload_asset("room1", tmp_path)
            assert result is None
            adapter._rpc.assert_not_called()
        finally:
            os.unlink(tmp_path)

    @pytest.mark.asyncio
    async def test_send_image_file_success(self):
        adapter = _make_adapter()

        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
            f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
            tmp_path = f.name

        try:
            rpc_responses = [
                (200, {"upload": {"id": "upload-1"}}),  # CreateUpload
                (200, {}),  # UploadChunk
                (200, {"asset": {"id": "asset-1"}}),  # CompleteUpload
                (200, {"message": {"id": "msg-1"}}),  # CreateMessage
            ]
            adapter._rpc = AsyncMock(side_effect=rpc_responses)

            result = await adapter.send_image_file("room1", tmp_path, caption="Test image")
            assert result.success is True
            assert result.message_id == "msg-1"

            # Verify the CreateMessage call had attachmentAssetIds
            create_msg_call = adapter._rpc.call_args_list[-1]
            body = create_msg_call.args[1]
            assert body["attachmentAssetIds"] == ["asset-1"]
            assert body["body"] == "Test image"
        finally:
            os.unlink(tmp_path)

    @pytest.mark.asyncio
    async def test_send_image_file_upload_fails_fallback(self):
        adapter = _make_adapter()

        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
            f.write(b"\x89PNG" + b"\x00" * 100)
            tmp_path = f.name

        try:
            # Upload fails, then fallback send succeeds
            rpc_responses = [
                (500, {"error": "fail"}),  # CreateUpload fails
                (200, {"message": {"id": "fallback-msg"}}),  # Fallback send
            ]
            adapter._rpc = AsyncMock(side_effect=rpc_responses)

            result = await adapter.send_image_file("room1", tmp_path)
            assert result.success is True
            assert result.message_id == "fallback-msg"
        finally:
            os.unlink(tmp_path)


# ── Seen tracking ─────────────────────────────────────────────────────────


class TestSeenTracking:
    """Test _mark_seen, _is_seen, and the _SEEN_CAP."""

    def test_mark_and_check_seen(self):
        adapter = _make_adapter()
        adapter._mark_seen("room1", "evt1")
        assert adapter._is_seen("room1", "evt1") is True
        assert adapter._is_seen("room1", "evt2") is False
        assert adapter._is_seen("room2", "evt1") is False

    def test_mark_seen_evicts_old_beyond_cap(self):
        adapter = _make_adapter()
        # Add more than _SEEN_CAP events
        for i in range(_SEEN_CAP + 10):
            adapter._mark_seen("room1", f"evt{i}")
        # The first events should have been evicted
        assert adapter._is_seen("room1", "evt0") is False
        # The most recent should still be there
        assert adapter._is_seen("room1", f"evt{_SEEN_CAP + 9}") is True
        # Total should not exceed cap
        assert len(adapter._seen["room1"]) <= _SEEN_CAP


# ── WebSocket URL building ────────────────────────────────────────────────


class TestWebSocketURL:
    """Test _websocket_url conversion."""

    def test_https_to_wss(self):
        adapter = _make_adapter()
        adapter._base_url = "https://chat.example.com"
        url = adapter._websocket_url()
        assert url.startswith("wss://")
        assert "/api/realtime" in url

    def test_http_to_ws(self):
        adapter = _make_adapter()
        adapter._base_url = "http://localhost:8080"
        url = adapter._websocket_url()
        assert url.startswith("ws://")
        assert "/api/realtime" in url

    def test_websocket_url_with_path(self):
        adapter = _make_adapter()
        adapter._base_url = "https://chat.example.com/subpath"
        url = adapter._websocket_url()
        assert "/subpath/api/realtime" in url

    def test_websocket_url_invalid_scheme(self):
        adapter = _make_adapter()
        adapter._base_url = "ftp://chat.example.com"
        with pytest.raises(ValueError, match="must use http"):
            adapter._websocket_url()


# ── Get chat info ─────────────────────────────────────────────────────────


class TestGetChatInfo:
    """Test get_chat_info."""

    @pytest.mark.asyncio
    async def test_get_chat_info_group(self):
        adapter = _make_adapter()
        adapter._room_names["room1"] = "General"
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        info = await adapter.get_chat_info("room1")
        assert info["name"] == "General"
        assert info["type"] == "group"

    @pytest.mark.asyncio
    async def test_get_chat_info_dm(self):
        adapter = _make_adapter()
        adapter._room_names["room2"] = "Alice"
        adapter._room_kinds["room2"] = "ROOM_KIND_DM"
        info = await adapter.get_chat_info("room2")
        assert info["name"] == "Alice"
        assert info["type"] == "dm"

    @pytest.mark.asyncio
    async def test_get_chat_info_unknown_room(self):
        adapter = _make_adapter()
        info = await adapter.get_chat_info("unknown-room")
        assert info["name"] == "unknown-room"
        assert info["type"] == "group"  # default


# ── Handle projection event ──────────────────────────────────────────────


class TestHandleProjectionEvent:
    """Test _handle_projection_event and _handle_timeline_event_upsert."""

    @pytest.mark.asyncio
    async def test_handle_projection_event_updates_cursor(self):
        adapter = _make_adapter()
        # Build a projection event with a resume cursor
        proj = _encode_field_string(4, "new-cursor-123")
        await adapter._handle_projection_event(proj)
        assert adapter._resume_cursor == "new-cursor-123"

    @pytest.mark.asyncio
    async def test_handle_projection_event_empty_data(self):
        adapter = _make_adapter()
        await adapter._handle_projection_event(b"")
        # Should not raise, cursor unchanged

    @pytest.mark.asyncio
    async def test_handle_timeline_event_dedup(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()
        adapter.handle_message = AsyncMock()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
        adapter._require_mention = False  # don't require mention for this test

        op = {
            "room_id": "room1",
            "event": {
                "id": "evt-dedup",
                "messagePosted": {
                    "message": {
                        "id": "m1",
                        "roomId": "room1",
                        "actorId": "user1",
                        "body": "Hello",
                        "createdAt": "",
                        "thread": {},
                    }
                },
            },
        }
        # First call should dispatch
        await adapter._handle_timeline_event_upsert(op)
        assert adapter.handle_message.called

        # Reset mock
        adapter.handle_message.reset_mock()

        # Second call should be deduped
        await adapter._handle_timeline_event_upsert(op)
        adapter.handle_message.assert_not_called()

    @pytest.mark.asyncio
    async def test_handle_timeline_event_no_message_posted(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()

        op = {
            "room_id": "room1",
            "event": {
                "id": "evt1",
                "messagePosted": None,
            },
        }
        await adapter._handle_timeline_event_upsert(op)
        adapter._message_handler.assert_not_called()

    @pytest.mark.asyncio
    async def test_handle_timeline_event_empty_event(self):
        adapter = _make_adapter()
        adapter._message_handler = AsyncMock()

        op = {"room_id": "room1", "event": {}}
        await adapter._handle_timeline_event_upsert(op)
        adapter._message_handler.assert_not_called()


# ── Handle transient event ────────────────────────────────────────────────


class TestHandleTransientEvent:
    """Test _handle_transient_event."""

    @pytest.mark.asyncio
    async def test_handle_transient_mention(self):
        adapter = _make_adapter()
        # Build a transient event envelope with a mention notification
        mention_inner = _encode_field_string(1, "room1") + _encode_field_string(2, "evt1")
        envelope = _encode_field_string(1, "env1") + _encode_submessage(88, mention_inner)
        # Should not raise
        await adapter._handle_transient_event(envelope)

    @pytest.mark.asyncio
    async def test_handle_transient_dm(self):
        adapter = _make_adapter()
        dm_inner = _encode_field_string(1, "room2") + _encode_field_string(2, "evt2")
        envelope = _encode_field_string(1, "env2") + _encode_submessage(89, dm_inner)
        await adapter._handle_transient_event(envelope)

    @pytest.mark.asyncio
    async def test_handle_transient_empty(self):
        adapter = _make_adapter()
        await adapter._handle_transient_event(b"")


# ── Disconnect ────────────────────────────────────────────────────────────


class TestDisconnect:
    """Test disconnect cleanup."""

    @pytest.mark.asyncio
    async def test_disconnect_cancels_typing_tasks(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._ws_active = False  # avoid websocket teardown
        adapter._liveness_task = None
        adapter._ws_task = None

        await adapter.send_typing("room1")
        assert "room1" in adapter._typing_tasks
        await adapter.disconnect()
        assert "room1" not in adapter._typing_tasks

    @pytest.mark.asyncio
    async def test_disconnect_clears_token(self):
        adapter = _make_adapter()
        adapter._rpc = AsyncMock(return_value=(200, {}))
        adapter._ws_active = False
        adapter._liveness_task = None
        adapter._ws_task = None
        assert adapter._token is not None
        await adapter.disconnect()
        assert adapter._token is None