signal-bot/tests/test_selftest.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Typing a command at the bot on its own behalf."""

from __future__ import annotations

import json
from types import SimpleNamespace
from typing import TYPE_CHECKING

import httpx
import pytest
from fastapi import Response, status

from signal_bot import app as app_module
from signal_bot.app import SelfTestRequest
from signal_bot.app import selftest as selftest_route
from signal_bot.bot import Bot, build_bot
from signal_bot.liveness import Liveness
from signal_bot.selftest import (
    SelfTestError,
    allowed_groups,
    build_frame,
    resolve_group,
    run,
)
from tests.conftest import WATCHED_GROUP, mock_client

if TYPE_CHECKING:
    from signal_bot.config import Config
    from signal_bot.discovery import GroupDirectory
    from signal_bot.store import Store
    from signal_bot.transport_fake import FakeTransport

GROUP_ID = "group.internal-id"
OTHER = "+48000000000"


def _bot(config: Config, store: Store, transport: FakeTransport, directory: GroupDirectory) -> Bot:
    """The production object graph, so !papież resolves the way it really does."""
    client = mock_client(_commons)
    return build_bot(config, client, store, transport, directory, Liveness(config, client))


def _commons(request: httpx.Request) -> httpx.Response:
    if "commons" in str(request.url):
        return httpx.Response(
            200,
            json={"query": {"pages": [{"imageinfo": [{"thumburl": "https://cdn/portrait.jpg"}]}]}},
        )
    return httpx.Response(200, content=b"portrait-bytes")


def _client(entries: list[dict[str, object]]) -> httpx.AsyncClient:
    def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=entries)

    return httpx.AsyncClient(transport=httpx.MockTransport(handler))


def test_no_groups_are_allowed_by_default() -> None:
    """Fail closed: a loopback POST must not reach the production group."""
    assert allowed_groups({}) == frozenset()


def test_the_allowlist_is_parsed_and_trimmed() -> None:
    assert allowed_groups({"SELFTEST_GROUPS": " testbot , catbot ,"}) == {"testbot", "catbot"}


def test_the_frame_is_the_shape_signal_cli_emits() -> None:
    body = json.loads(build_frame(GROUP_ID, OTHER, "!papież"))["envelope"]
    assert body["sourceUuid"] == OTHER
    assert body["dataMessage"]["message"] == "!papież"
    assert body["dataMessage"]["groupInfo"] == {"groupId": GROUP_ID, "type": "DELIVER"}
    assert body["timestamp"] == body["dataMessage"]["timestamp"]


async def test_resolve_group_picks_a_member_who_is_not_the_bot(config: Config) -> None:
    """The reply quotes the author, so it has to be somebody Signal knows."""
    entries = [
        {"name": "testbot", "internal_id": GROUP_ID, "members": [config.phone_number, OTHER]}
    ]
    assert await resolve_group(_client(entries), config, "testbot") == (GROUP_ID, OTHER)


async def test_resolve_group_rejects_an_unknown_name(config: Config) -> None:
    with pytest.raises(SelfTestError, match="no group named"):
        await resolve_group(_client([]), config, "nope")


async def test_resolve_group_rejects_a_group_with_nobody_else_in_it(config: Config) -> None:
    entries = [{"name": "testbot", "internal_id": GROUP_ID, "members": [config.phone_number]}]
    with pytest.raises(SelfTestError, match="no usable id or no other member"):
        await resolve_group(_client(entries), config, "testbot")


async def test_run_refuses_a_group_outside_the_allowlist(
    config: Config, store: Store, transport: FakeTransport, directory: GroupDirectory
) -> None:
    bot = _bot(config, store, transport, directory)
    with pytest.raises(SelfTestError, match="refusing to post"):
        await run(bot, _client([]), "realgroup", "!papiez", env={"SELFTEST_GROUPS": "testbot"})


async def test_run_feeds_the_live_dispatcher(
    config: Config, store: Store, transport: FakeTransport, directory: GroupDirectory
) -> None:
    """The point of the module: the real wiring, not a reconstruction.

    build_bot is what production calls, so a command that resolves here is a
    command the group can actually type -- which is exactly what could not
    be checked before, Signal never delivering an account its own sends.
    """
    bot = _bot(config, store, transport, directory)
    entries = [
        {"name": "testbot", "internal_id": WATCHED_GROUP, "members": [config.phone_number, OTHER]}
    ]

    delivered = await run(
        bot, _client(entries), "testbot", "!papiez", env={"SELFTEST_GROUPS": "testbot"}
    )

    assert delivered == WATCHED_GROUP
    assert transport.sent[0].image_base64 is not None
    assert "pl.wikipedia.org" in transport.sent[0].caption


async def test_the_endpoint_reports_connecting_before_the_bot_is_up() -> None:
    """A self-test during startup is not a failure, it is "not yet"."""
    app_module.app.state.starter = SimpleNamespace(runtime=None, reason="container down")
    response = Response()

    body = await selftest_route(SelfTestRequest(group="testbot", text="!papiez"), response)

    assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
    assert body == {"status": "connecting", "reason": "container down"}


async def test_the_endpoint_refuses_a_group_outside_the_allowlist(
    config: Config, store: Store, transport: FakeTransport, directory: GroupDirectory
) -> None:
    bot = _bot(config, store, transport, directory)
    app_module.app.state.starter = SimpleNamespace(runtime=SimpleNamespace(bot=bot), reason="")
    app_module.app.state.client = _client([])
    response = Response()

    body = await selftest_route(SelfTestRequest(group="realgroup", text="!papiez"), response)

    assert response.status_code == status.HTTP_403_FORBIDDEN
    assert body["status"] == "refused"
    assert transport.sent == []


async def test_the_endpoint_delivers_and_says_where(
    config: Config,
    store: Store,
    transport: FakeTransport,
    directory: GroupDirectory,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("SELFTEST_GROUPS", "testbot")
    bot = _bot(config, store, transport, directory)
    app_module.app.state.starter = SimpleNamespace(runtime=SimpleNamespace(bot=bot), reason="")
    app_module.app.state.client = _client(
        [{"name": "testbot", "internal_id": WATCHED_GROUP, "members": [config.phone_number, OTHER]}]
    )
    response = Response()

    body = await selftest_route(SelfTestRequest(group="testbot", text="!papiez"), response)

    assert body == {"status": "delivered", "group": WATCHED_GROUP, "text": "!papiez"}
    assert transport.sent[0].image_base64 is not None


async def test_resolve_group_rejects_an_entry_with_no_id(config: Config) -> None:
    entries = [{"name": "testbot", "members": [config.phone_number, OTHER]}]
    with pytest.raises(SelfTestError, match="no usable id or no other member"):
        await resolve_group(_client(entries), config, "testbot")


async def test_resolve_group_skips_past_groups_with_other_names(config: Config) -> None:
    """The bot is in several groups; only the named one may be posted into."""
    entries = [
        {"name": "produkcja", "internal_id": "group.real", "members": [config.phone_number, OTHER]},
        {"name": "testbot", "internal_id": GROUP_ID, "members": [config.phone_number, OTHER]},
    ]
    assert await resolve_group(_client(entries), config, "testbot") == (GROUP_ID, OTHER)