signal-bot/tests/test_discovery.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The live group set: discovery, the allowlist, and never narrowing on failure."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

import httpx
import pytest

from signal_bot.discovery import GroupDirectory
from signal_bot.stats.events import MessageEvent
from signal_bot.trust import InviteTrust
from tests.conftest import (
    OTHER_GROUP,
    SECOND_GROUP,
    WATCHED_GROUP,
    WATCHED_INTERNAL,
    group_entry,
    listing_client,
)

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

Listing = list[dict[str, Any]]


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


async def test_discovers_every_group_the_account_is_in(config: Config, store: Store) -> None:
    """No GROUP_IDS at all: whatever the account is a member of is watched."""
    found = GroupDirectory(
        config,
        listing_client([group_entry(WATCHED_GROUP), group_entry(SECOND_GROUP)]),
        store,
        InviteTrust(()),
    )
    await found.refresh()
    assert found.groups == (WATCHED_GROUP, SECOND_GROUP)
    assert found.resolve(WATCHED_INTERNAL) == WATCHED_GROUP
    assert found.resolve(OTHER_GROUP) is None
    assert found.resolve(None) is None


async def test_a_group_that_is_only_an_invite_is_not_watched(config: Config, store: Store) -> None:
    """Joining is attempted, but the listing decides whether it worked."""
    found = _directory(
        config,
        store,
        lambda r: httpx.Response(200, json=[group_entry(WATCHED_GROUP, member=False)]),
    )
    await found.refresh()
    assert found.groups == ()


async def test_a_pending_invitation_is_joined(config: Config, store: Store) -> None:
    # Two GETs per refresh when a join is attempted: read, join, re-read.
    # The re-read is the authority -- the join endpoint answers 201 either way.
    listings = iter(
        [
            [group_entry(WATCHED_GROUP, member=False)],
            [group_entry(WATCHED_GROUP)],
        ]
    )

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

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


async def test_an_impossible_invitation_is_not_retried_forever(
    config: Config, store: Store
) -> None:
    """signal-cli cannot accept an invite addressed to a phone number."""
    posts: list[str] = []

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

    found = _directory(config, store, handler)
    await found.refresh()
    await found.refresh()
    await found.refresh()
    assert len(posts) == 1


async def test_a_failed_listing_never_narrows_the_watched_set(config: Config, store: Store) -> None:
    """The worst bug available: conclude you were removed from everything."""
    responses = iter(
        [
            httpx.Response(200, json=[group_entry(WATCHED_GROUP)]),
            httpx.Response(500, text="upstream is down"),
        ]
    )
    found = _directory(config, store, lambda r: next(responses))
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)
    assert found.resolve(WATCHED_INTERNAL) == WATCHED_GROUP


async def test_a_transport_error_never_narrows_the_watched_set(
    config: Config, store: Store
) -> None:
    calls = {"n": 0}

    def handler(request: httpx.Request) -> httpx.Response:
        calls["n"] += 1
        if calls["n"] == 1:
            return httpx.Response(200, json=[group_entry(WATCHED_GROUP)])
        msg = "signal-cli went away"
        raise httpx.ConnectError(msg)

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


@pytest.mark.parametrize("body", [{"error": "nope"}, "a string"])
async def test_a_listing_that_is_not_a_list_is_ignored(
    config: Config, store: Store, body: object
) -> None:
    responses = iter(
        [httpx.Response(200, json=[group_entry(WATCHED_GROUP)]), httpx.Response(200, json=body)]
    )
    found = _directory(config, store, lambda r: next(responses))
    await found.refresh()
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)


async def test_rows_that_are_not_objects_are_skipped(config: Config, store: Store) -> None:
    found = _directory(
        config, store, lambda r: httpx.Response(200, json=["nonsense", group_entry(WATCHED_GROUP)])
    )
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)


@pytest.mark.parametrize("entry", [{"member": True}, {"id": "", "member": True}, {"id": 7}])
async def test_a_row_without_a_usable_id_is_skipped(
    config: Config, store: Store, entry: dict[str, Any]
) -> None:
    found = _directory(config, store, lambda r: httpx.Response(200, json=[entry]))
    await found.refresh()
    assert found.groups == ()


async def test_a_removed_group_is_dropped_but_keeps_its_history(
    config: Config, store: Store
) -> None:
    """Leaving clears the live scaffolding and keeps the statistics.

    Before 0.1.0 this purged the group's counts. It cannot any more: the same
    code path fires when a listing transiently omits members, and losing years
    of history to one bad HTTP response is not a trade worth making.
    """
    responses = iter(
        [
            httpx.Response(200, json=[group_entry(WATCHED_GROUP), group_entry(SECOND_GROUP)]),
            httpx.Response(200, json=[group_entry(SECOND_GROUP)]),
        ]
    )
    found = _directory(config, store, lambda r: next(responses))
    await found.refresh()
    store.record_use(WATCHED_GROUP, "u1", "cat", 100.0, "2026-09-01")
    store.events.record(
        MessageEvent(
            group_id=WATCHED_GROUP,
            uuid="u1",
            display_name="Alice",
            sent_at_ms=1,
            day="2026-09-01",
            local_hour=12,
        )
    )
    store.record_use(SECOND_GROUP, "u2", "cat", 200.0, "2026-09-01")

    await found.refresh()

    assert found.groups == (SECOND_GROUP,)
    assert found.resolve(WATCHED_INTERNAL) is None
    assert store.last_used(WATCHED_GROUP, "u1", "cat") is None
    kept = store.events.connection.execute(
        "SELECT COUNT(*) AS n FROM message_events WHERE group_id = ?", (WATCHED_GROUP,)
    ).fetchone()
    assert kept["n"] == 1
    assert store.last_used(SECOND_GROUP, "u2", "cat") == 200.0


async def test_refresh_for_looks_again_once_per_unknown_group(config: Config, store: Store) -> None:
    """The first command after an add must work, without hammering the API."""
    listings: list[Any] = [
        [],
        [group_entry(WATCHED_GROUP)],
        [group_entry(WATCHED_GROUP)],
    ]
    calls = {"n": 0}

    def handler(request: httpx.Request) -> httpx.Response:
        index = min(calls["n"], len(listings) - 1)
        calls["n"] += 1
        return httpx.Response(200, json=listings[index])

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

    assert await found.refresh_for(WATCHED_INTERNAL) == WATCHED_GROUP
    before = calls["n"]
    # A second unknown id refreshes; the same one again does not.
    assert await found.refresh_for(OTHER_GROUP) is None
    assert calls["n"] > before
    after = calls["n"]
    assert await found.refresh_for(OTHER_GROUP) is None
    assert calls["n"] == after