signal-bot/tests/test_transport_rest.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The bbernhard adapter: sends, attachments, mode check and reconnect."""

from __future__ import annotations

from collections.abc import AsyncIterator, Callable
from typing import TYPE_CHECKING

import httpx
import pytest
from websockets.exceptions import ConnectionClosedError

from signal_bot.transport_rest import RestTransport, TransportError

if TYPE_CHECKING:
    from types import TracebackType

    from signal_bot.config import Config

# The callable httpx.MockTransport expects -- not a bot command handler.
MockHandler = Callable[[httpx.Request], httpx.Response]


class FakeSocket:
    """Yields a canned list of frames, optionally failing partway through."""

    def __init__(self, frames: list[str], *, fail_after: bool = False) -> None:
        self._frames = frames
        self._fail_after = fail_after

    async def __aiter__(self) -> AsyncIterator[str]:
        for frame in self._frames:
            yield frame
        if self._fail_after:
            raise ConnectionClosedError(None, None)


class FakeConnect:
    """Stands in for websockets.connect, handing out sockets in order."""

    def __init__(self, sockets: list[FakeSocket]) -> None:
        self._sockets = sockets
        self.urls: list[str] = []

    def __call__(self, url: str) -> FakeConnect:
        self.urls.append(url)
        return self

    async def __aenter__(self) -> FakeSocket:
        if not self._sockets:
            msg = "no more sockets"
            raise OSError(msg)
        return self._sockets.pop(0)

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        return None


def _transport(config: Config, handler: MockHandler, **kwargs: object) -> RestTransport:
    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    return RestTransport(config, client, **kwargs)


async def test_send_text_posts_expected_payload(config: Config) -> None:
    seen: list[dict] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(__import__("json").loads(request.content))
        return httpx.Response(201)

    await _transport(config, handler).send_text("hi", "group==")
    assert seen[0] == {
        "message": "hi",
        "number": config.phone_number,
        "recipients": ["group=="],
    }


async def test_send_image_posts_base64_attachment(config: Config) -> None:
    seen: list[dict] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(__import__("json").loads(request.content))
        return httpx.Response(200)

    await _transport(config, handler).send_image("BASE64", "group==")
    assert seen[0]["base64_attachments"] == ["BASE64"]


async def test_send_raises_on_rejection(config: Config) -> None:
    transport = _transport(config, lambda r: httpx.Response(400, text="nope"))
    with pytest.raises(TransportError, match="400"):
        await transport.send_text("hi", "g")


async def test_verify_mode_accepts_json_rpc(config: Config) -> None:
    transport = _transport(config, lambda r: httpx.Response(200, json={"mode": "json-rpc"}))
    assert await transport.verify_mode() == "json-rpc"


async def test_verify_mode_rejects_the_native_build(config: Config) -> None:
    """It serves the same WebSocket and then fails every attachment send."""
    transport = _transport(config, lambda r: httpx.Response(200, json={"mode": "json-rpc-native"}))
    with pytest.raises(TransportError, match="awt"):
        await transport.verify_mode()


@pytest.mark.parametrize("body", [{"mode": "normal"}, {"mode": "native"}, {}, ["nope"]])
async def test_verify_mode_rejects_non_websocket_modes(config: Config, body: object) -> None:
    """In normal/native mode /v1/receive silently degrades to a polling GET."""
    transport = _transport(config, lambda r: httpx.Response(200, json=body))
    with pytest.raises(TransportError, match="not 'json-rpc'"):
        await transport.verify_mode()


async def test_list_attachments(config: Config) -> None:
    transport = _transport(config, lambda r: httpx.Response(200, json=["a", "b"]))
    assert await transport.list_attachments() == ["a", "b"]


async def test_list_attachments_tolerates_non_list(config: Config) -> None:
    transport = _transport(config, lambda r: httpx.Response(200, json={"oops": 1}))
    assert await transport.list_attachments() == []


async def test_delete_attachment(config: Config) -> None:
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(str(request.url))
        return httpx.Response(204)

    await _transport(config, handler).delete_attachment("abc")
    assert seen[0].endswith("/v1/attachments/abc")


async def test_delete_attachment_raises_on_failure(config: Config) -> None:
    transport = _transport(config, lambda r: httpx.Response(500))
    with pytest.raises(TransportError, match="500"):
        await transport.delete_attachment("abc")


async def test_receive_yields_frames(config: Config) -> None:
    connector = FakeConnect([FakeSocket(["a", "b"])])
    transport = _transport(
        config,
        lambda r: httpx.Response(200),
        connector=connector,
        sleeper=_no_sleep,
        max_reconnects=0,
    )
    assert [f async for f in transport.receive()] == ["a", "b"]


async def test_receive_decodes_binary_frames(config: Config) -> None:
    connector = FakeConnect([FakeSocket([b"bytes-frame"])])
    transport = _transport(
        config,
        lambda r: httpx.Response(200),
        connector=connector,
        sleeper=_no_sleep,
        max_reconnects=0,
    )
    assert [f async for f in transport.receive()] == ["bytes-frame"]


async def test_receive_reconnects_after_a_drop(config: Config) -> None:
    """A container restart used to kill the bot silently."""
    slept: list[float] = []

    async def sleeper(seconds: float) -> None:
        slept.append(seconds)

    connector = FakeConnect([FakeSocket(["a"], fail_after=True), FakeSocket(["b"])])
    transport = _transport(
        config,
        lambda r: httpx.Response(200),
        connector=connector,
        sleeper=sleeper,
        max_reconnects=1,
    )
    assert [f async for f in transport.receive()] == ["a", "b"]
    assert slept == [1.0]


async def test_receive_backs_off_exponentially(config: Config) -> None:
    slept: list[float] = []

    async def sleeper(seconds: float) -> None:
        slept.append(seconds)

    connector = FakeConnect([])
    transport = _transport(
        config,
        lambda r: httpx.Response(200),
        connector=connector,
        sleeper=sleeper,
        max_reconnects=3,
    )
    assert [f async for f in transport.receive()] == []
    assert slept == [1.0, 2.0, 4.0]


async def _no_sleep(seconds: float) -> None:
    return None


def _groups(*entries: object) -> MockHandler:
    """Answer /v1/groups with an arbitrary payload."""
    return lambda r: httpx.Response(200, json=list(entries))


async def test_membership_accepts_a_group_matched_by_its_internal_id(
    config: Config,
) -> None:
    """Envelopes carry internal_id, so that spelling must satisfy the gate."""
    handler = _groups({"id": "group.other", "internal_id": config.group_id, "member": True})
    assert await _transport(config, handler).verify_membership() is None


async def test_membership_rejects_a_group_the_bot_was_only_invited_to(
    config: Config,
) -> None:
    handler = _groups({"id": config.group_id, "name": "testbot", "member": False})
    with pytest.raises(TransportError, match="has not joined"):
        await _transport(config, handler).verify_membership()


async def test_membership_rejects_an_account_in_no_matching_group(
    config: Config,
) -> None:
    handler = _groups({"id": "group.someone-elses", "member": True})
    with pytest.raises(TransportError, match="in no group matching"):
        await _transport(config, handler).verify_membership()


async def test_membership_ignores_entries_that_are_not_objects(
    config: Config,
) -> None:
    """A malformed row must not mask the real reason the bot is deaf."""
    handler = _groups("unexpected", {"id": config.group_id, "member": True})
    assert await _transport(config, handler).verify_membership() is None