opensquilla--opensquilla/tests/test_channels/test_slack_socket_mode.py

"""Slack Socket Mode transport, auto-target replies, and self-echo filtering."""

from __future__ import annotations

import asyncio
from typing import Any

import pytest

from opensquilla.channels.manager import ChannelManager
from opensquilla.channels.slack import SlackAuthError, SlackChannel
from opensquilla.channels.types import IncomingMessage, OutgoingMessage


def _mk(**kwargs: Any) -> SlackChannel:
    kwargs.setdefault("slack_channel_id", "")
    ch = SlackChannel(token="xoxb-test", **kwargs)
    ch.bot_user_id = "UBOT"
    return ch


class _FakeResp:
    def __init__(self, payload: dict[str, Any]) -> None:
        self._payload = payload

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, Any]:
        return self._payload


class _FakeClient:
    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, Any] | None, Any]] = []
        self.socket_open_payload: dict[str, Any] = {
            "ok": True,
            "url": "wss://socket.slack.test/session",
        }

    async def post(
        self, path: str, json: dict[str, Any] | None = None, headers: Any = None
    ) -> _FakeResp:
        self.calls.append((path, json, headers))
        if path == "/auth.test":
            return _FakeResp({"ok": True, "user_id": "UBOT"})
        if path == "/apps.connections.open":
            return _FakeResp(self.socket_open_payload)
        return _FakeResp({"ok": True, "ts": "1700000000.000100"})


def test_transport_name_follows_connection_mode() -> None:
    assert _mk().transport_name == "webhook"
    assert _mk(connection_mode="socket").transport_name == "websocket"


async def test_socket_mode_requires_app_token() -> None:
    ch = _mk(connection_mode="socket")  # no app_token
    ch._get_client = lambda: _FakeClient()  # type: ignore[method-assign]
    with pytest.raises(SlackAuthError):
        await ch.start()


async def test_socket_mode_validates_app_token_before_reporting_started() -> None:
    ch = _mk(connection_mode="socket", app_token="xapp-valid")
    fake = _FakeClient()
    fake.socket_open_payload = {"ok": False, "error": "invalid_auth"}
    ch._get_client = lambda: fake  # type: ignore[method-assign]
    with pytest.raises(SlackAuthError, match="apps.connections.open failed"):
        await ch.start()
    assert ch.is_connected() is False


async def test_socket_mode_start_passes_opened_url_to_background_loop() -> None:
    ch = _mk(connection_mode="socket", app_token="xapp-valid")
    fake = _FakeClient()
    seen: list[str | None] = []

    async def _fake_socket_loop(initial_socket_url: str | None = None) -> None:
        seen.append(initial_socket_url)
        await ch._socket_stop.wait()  # type: ignore[union-attr]

    ch._get_client = lambda: fake  # type: ignore[method-assign]
    ch._run_socket_loop = _fake_socket_loop  # type: ignore[method-assign]

    await ch.start()
    await asyncio.sleep(0)
    await ch.stop()

    assert seen == ["wss://socket.slack.test/session"]
    open_call = next(c for c in fake.calls if c[0] == "/apps.connections.open")
    assert open_call[2] == {"Authorization": "Bearer xapp-valid"}


def test_ingest_accepts_plain_user_message() -> None:
    ch = _mk()
    ch._ingest_event_callback(
        {
            "event_id": "Ev1",
            "event": {
                "type": "message",
                "user": "UUSER",
                "channel": "D123",
                "text": "hi",
                "ts": "1.1",
            },
        }
    )
    assert ch._queue.qsize() == 1
    msg = ch._queue.get_nowait()
    assert msg.channel_id == "D123"
    assert msg.content == "hi"


def test_ingest_accepts_app_mention_event() -> None:
    ch = _mk()
    ch._ingest_event_callback(
        {
            "event_id": "EvMention",
            "event": {
                "type": "app_mention",
                "user": "UUSER",
                "channel": "C123",
                "text": "<@UBOT> hi",
                "ts": "2.1",
            },
        }
    )

    assert ch._queue.qsize() == 1
    msg = ch._queue.get_nowait()
    assert msg.channel_id == "C123"
    assert msg.content == "<@UBOT> hi"


@pytest.mark.parametrize(
    "event",
    [
        {"type": "message", "bot_id": "B1", "channel": "D1", "text": "x", "ts": "2"},
        {"type": "message", "user": "UBOT", "channel": "D1", "text": "x", "ts": "3"},
        {"type": "message", "subtype": "bot_message", "channel": "D1", "text": "x", "ts": "4"},
        {"type": "message", "subtype": "message_changed", "channel": "D1", "ts": "5"},
        {"type": "message", "subtype": "message_deleted", "channel": "D1", "ts": "6"},
    ],
)
def test_ingest_drops_self_echoes_and_non_user_subtypes(event: dict[str, Any]) -> None:
    ch = _mk()
    ch._ingest_event_callback({"event_id": f"e-{event.get('ts')}", "event": event})
    assert ch._queue.qsize() == 0


def test_ingest_dedupes_replayed_event() -> None:
    ch = _mk()
    payload = {
        "event_id": "Dup1",
        "event": {"type": "message", "user": "U", "channel": "D1", "text": "hi", "ts": "9"},
    }
    ch._ingest_event_callback(payload)
    ch._ingest_event_callback(payload)
    assert ch._queue.qsize() == 1


def test_ingest_dedupes_app_mention_and_message_pair() -> None:
    ch = _mk()
    ch._ingest_event_callback(
        {
            "event_id": "EvMention",
            "event": {
                "type": "app_mention",
                "user": "UUSER",
                "channel": "C123",
                "text": "<@UBOT> hi",
                "ts": "10.1",
            },
        }
    )
    ch._ingest_event_callback(
        {
            "event_id": "EvMessage",
            "event": {
                "type": "message",
                "user": "UUSER",
                "channel": "C123",
                "text": "<@UBOT> hi",
                "ts": "10.1",
            },
        }
    )

    assert ch._queue.qsize() == 1


class _FakeSocket:
    def __init__(self) -> None:
        self.sent: list[str] = []
        self.closed = False

    async def send(self, payload: str) -> None:
        self.sent.append(payload)

    async def close(self) -> None:
        self.closed = True


async def test_socket_frame_acks_and_dispatches_events_api_payload() -> None:
    ch = _mk()
    ws = _FakeSocket()
    await ch._handle_socket_frame(
        ws,
        (
            '{"envelope_id":"env-1","type":"events_api","payload":'
            '{"type":"event_callback","event_id":"Ev1","event":'
            '{"type":"message","user":"UUSER","channel":"D123","text":"hi","ts":"1.1"}}}'
        ),
    )

    assert ws.sent == ['{"envelope_id": "env-1"}']
    assert ch._queue.qsize() == 1


async def test_socket_frame_disconnect_closes_socket_after_ack() -> None:
    ch = _mk()
    ws = _FakeSocket()
    await ch._handle_socket_frame(ws, '{"envelope_id":"env-2","type":"disconnect"}')

    assert ws.sent == ['{"envelope_id": "env-2"}']
    assert ws.closed is True


async def test_send_auto_targets_reply_conversation() -> None:
    ch = _mk()  # slack_channel_id empty on purpose
    fake = _FakeClient()
    ch._get_client = lambda: fake  # type: ignore[method-assign]
    await ch.send(OutgoingMessage(content="hello", reply_to="D999"))
    post = next(c for c in fake.calls if c[0] == "/chat.postMessage")
    assert post[1] is not None
    assert post[1]["channel"] == "D999"
    # A conversation id must NOT be misused as a thread anchor.
    assert "thread_ts" not in post[1]


async def test_send_threads_only_on_message_ts() -> None:
    ch = _mk(slack_channel_id="C1")
    fake = _FakeClient()
    ch._get_client = lambda: fake  # type: ignore[method-assign]
    await ch.send(OutgoingMessage(content="hi", reply_to="1700000000.000200"))
    post = next(c for c in fake.calls if c[0] == "/chat.postMessage")
    assert post[1] is not None
    assert post[1]["channel"] == "C1"
    assert post[1]["thread_ts"] == "1700000000.000200"


async def test_send_thread_timestamp_uses_metadata_channel_without_default() -> None:
    ch = _mk()
    fake = _FakeClient()
    ch._get_client = lambda: fake  # type: ignore[method-assign]
    await ch.send(
        OutgoingMessage(
            content="hi",
            reply_to="1700000000.000200",
            metadata={"channel": "C42"},
        )
    )
    post = next(c for c in fake.calls if c[0] == "/chat.postMessage")
    assert post[1] is not None
    assert post[1]["channel"] == "C42"
    assert post[1]["thread_ts"] == "1700000000.000200"


def test_reply_helpers_target_inbound_conversation() -> None:
    ch = _mk()
    inbound = IncomingMessage(sender_id="U", channel_id="D42", content="hi")
    assert ch.build_reply_message("r", inbound).reply_to == "D42"
    assert ch.build_reply_message("r", inbound).metadata == {"channel": "D42"}
    assert ch.streaming_reply_kwargs(inbound) == {"channel": "D42"}


def test_reply_helpers_preserve_thread_target_when_enabled() -> None:
    ch = _mk(reply_in_thread=True)
    inbound = IncomingMessage(
        sender_id="U",
        channel_id="C42",
        content="hi",
        metadata={"ts": "1700000000.000200", "thread_ts": "1700000000.000100"},
    )

    reply = ch.build_reply_message("r", inbound)

    assert reply.reply_to == "C42"
    assert reply.metadata == {
        "channel": "C42",
        "thread_ts": "1700000000.000100",
    }
    assert ch.streaming_reply_kwargs(inbound) == {
        "channel": "C42",
        "thread_ts": "1700000000.000100",
    }


def test_reply_helpers_thread_root_when_enabled() -> None:
    ch = _mk(reply_in_thread=True)
    inbound = IncomingMessage(
        sender_id="U",
        channel_id="C42",
        content="hi",
        metadata={"ts": "1700000000.000200"},
    )

    assert ch.build_reply_message("r", inbound).metadata == {
        "channel": "C42",
        "thread_ts": "1700000000.000200"
    }


def test_channel_manager_skips_webhook_route_for_slack_socket_mode() -> None:
    manager = ChannelManager(
        _channels={
            "webhook": _mk(connection_mode="webhook"),
            "socket": _mk(connection_mode="socket", app_token="xapp-valid"),
        },
        _turn_runner=None,
        _session_manager=None,
    )

    routes = manager.collect_webhook_routes()

    assert [route.path for route in routes] == ["/slack/events"]