signal-bot/tests/test_allowlist.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""GROUP_IDS as a restriction: pinning the bot, and refusing to start unpinned."""

from __future__ import annotations

from dataclasses import replace
from typing import TYPE_CHECKING

import httpx
import pytest

from signal_bot.discovery import GroupDirectory
from signal_bot.membership import MembershipError
from signal_bot.trust import InviteTrust
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, group_entry

if TYPE_CHECKING:
    from signal_bot.config import Config
    from signal_bot.store import Store


def _directory(config: Config, store: Store, handler: object) -> GroupDirectory:
    return GroupDirectory(
        config,
        httpx.AsyncClient(transport=httpx.MockTransport(handler)),
        store,
        InviteTrust(()),
    )


async def test_the_allowlist_restricts_what_is_adopted(config: Config, store: Store) -> None:
    """GROUP_IDS still pins the bot when somebody wants it pinned."""
    pinned = replace(config, group_ids=(WATCHED_GROUP,))
    found = _directory(
        pinned,
        store,
        lambda r: httpx.Response(200, json=[group_entry(WATCHED_GROUP), group_entry(SECOND_GROUP)]),
    )
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)
    await found.verify_allowlist()


async def test_verify_allowlist_names_a_pinned_group_that_is_missing(
    config: Config, store: Store
) -> None:
    pinned = replace(config, group_ids=(WATCHED_GROUP, SECOND_GROUP))
    found = _directory(
        pinned, store, lambda r: httpx.Response(200, json=[group_entry(WATCHED_GROUP)])
    )
    await found.refresh()
    with pytest.raises(MembershipError, match="not a joined member") as raised:
        await found.verify_allowlist()
    assert SECOND_GROUP in str(raised.value)


async def test_verify_allowlist_reports_why_the_join_failed(config: Config, store: Store) -> None:
    """The refusal carries the API's own answer, not a generic complaint."""
    pinned = replace(config, group_ids=(WATCHED_GROUP,))

    def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "POST":
            return httpx.Response(400, text="Cannot find service ID for self")
        return httpx.Response(200, json=[group_entry(WATCHED_GROUP, member=False)])

    found = _directory(pinned, store, handler)
    await found.refresh()
    with pytest.raises(MembershipError, match="Cannot find service ID"):
        await found.verify_allowlist()


async def test_an_unpinned_bot_in_no_groups_is_allowed_to_start(
    config: Config, store: Store
) -> None:
    found = _directory(config, store, lambda r: httpx.Response(200, json=[]))
    await found.refresh()
    await found.verify_allowlist()


async def test_run_refreshes_on_a_schedule(config: Config, store: Store) -> None:
    """A group added while nobody is talking is still picked up."""
    slept: list[float] = []
    listings = iter([[], [group_entry(WATCHED_GROUP)]])

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

    found = GroupDirectory(
        config,
        httpx.AsyncClient(
            transport=httpx.MockTransport(lambda r: httpx.Response(200, json=next(listings)))
        ),
        store,
        InviteTrust(()),
        sleeper=sleeper,
    )
    await found.run(iterations=2)
    assert slept == [30.0, 30.0]
    assert found.groups == (WATCHED_GROUP,)


async def test_a_failed_re_read_after_joining_keeps_the_old_set(
    config: Config, store: Store
) -> None:
    """Fail closed: an unverified join must not be treated as membership."""
    responses = iter(
        [
            httpx.Response(200, json=[group_entry(WATCHED_GROUP)]),
            httpx.Response(
                200, json=[group_entry(WATCHED_GROUP), group_entry(SECOND_GROUP, member=False)]
            ),
            httpx.Response(201, text=""),
            httpx.Response(503, text="gone"),
        ]
    )

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(201, text="") if request.method == "POST" else next(responses)

    found = _directory(config, store, handler)
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)