signal-bot/tests/conftest.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Shared fixtures and envelope builders."""

from __future__ import annotations

import datetime as dt
import json
from typing import TYPE_CHECKING, Any

import httpx
import pytest

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
from signal_bot.trust import InviteTrust

if TYPE_CHECKING:
    from collections.abc import Callable
    from pathlib import Path

# Canonical ``group.<base64>`` ids, the spelling /v1/groups reports and the
# only one /v2/send accepts. The *_INTERNAL forms are what an envelope
# carries: exactly what the canonical form decodes to.
WATCHED_GROUP = "group.ZDJGMFkyaGxaQzFuY205MWNDMXBaQzB6TWkxaWVYUmxjeUVoSVNFaElRPT0="
WATCHED_INTERNAL = "d2F0Y2hlZC1ncm91cC1pZC0zMi1ieXRlcyEhISEhIQ=="
SECOND_GROUP = "group.YzJWamIyNWtMV2R5YjNWd0xXbGtMVE15TFdKNWRHVnpJU0VoSVNFaA=="
SECOND_INTERNAL = "c2Vjb25kLWdyb3VwLWlkLTMyLWJ5dGVzISEhISEh"
OTHER_GROUP = "group.YjNSb1pYSXRaM0p2ZFhBdGFXUXRNekl0WW5sMFpYTWhJU0VoSVNFPQ=="

# The operator, and somebody who is not.
TRUSTED_NUMBER = "+48123456789"
STRANGER_NUMBER = "+15555550123"


@pytest.fixture
def config() -> Config:
    """The production shape: no allowlist, groups discovered from the account."""
    return Config(
        phone_number="+15550000000",
        api_base_url="http://localhost:9922",
        database_path=":memory:",
        cooldown_seconds=10,
    )


def group_entry(group_id: str, *, member: bool = True, name: str = "e2e") -> dict[str, Any]:
    """One /v1/groups row, in both spellings the real API uses."""
    return {"id": group_id, "internal_id": group_id, "name": name, "member": member}


def listing_client(entries: list[dict[str, Any]]) -> httpx.AsyncClient:
    """A client whose /v1/groups always answers with these rows."""
    return mock_client(lambda request: httpx.Response(200, json=entries))


# Captured before any monkeypatching, so a test that patches
# httpx.AsyncClient cannot make this helper recurse into itself.
_REAL_ASYNC_CLIENT = httpx.AsyncClient


def mock_client(handler: Callable[[httpx.Request], httpx.Response]) -> httpx.AsyncClient:
    """A client whose every request is answered by this handler."""
    return _REAL_ASYNC_CLIENT(transport=httpx.MockTransport(handler))


def signal_client(
    *group_ids: str, mode: str = "json-rpc", member: bool = True
) -> httpx.AsyncClient:
    """Answer both startup probes: the container mode, then the group listing."""

    def handler(request: httpx.Request) -> httpx.Response:
        if "/v1/groups" in request.url.path:
            return httpx.Response(200, json=[group_entry(gid, member=member) for gid in group_ids])
        return httpx.Response(200, json={"mode": mode})

    return mock_client(handler)


@pytest.fixture
async def directory(config: Config, store: Store) -> GroupDirectory:
    """A directory that has already discovered both test groups."""
    client = listing_client([group_entry(WATCHED_GROUP), group_entry(SECOND_GROUP)])
    found = GroupDirectory(config, client, store, InviteTrust(()))
    await found.refresh()
    return found


@pytest.fixture(autouse=True)
def _portraits_stay_out_of_the_real_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Point the portrait cache at a temp dir for every test in the suite.

    Without this, any test that runs the !papież handler writes into the
    caller's real ~/.cache/signal-bot/portraits -- silently, and while still
    passing. Autouse because the write happens several layers below the test
    that triggers it, so opting in per test is a rule nobody would remember.
    """
    monkeypatch.setenv("PORTRAIT_CACHE_DIR", str(tmp_path / "portraits"))


@pytest.fixture(autouse=True)
def _no_test_ever_reads_a_real_photo_pack(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Keep the whole suite away from the operator's private album.

    The pack directory is discovered, not configured: with nothing set,
    signal_bot.pack.default_dir falls back to $XDG_DATA_HOME and then to
    $HOME, so a developer who has an album would have the suite load it --
    quietly, and while still passing. Pointing XDG_DATA_HOME at an empty
    temp dir makes "no pack" the answer everywhere unless a test builds one.
    """
    monkeypatch.delenv("PHOTO_PACK_DIR", raising=False)
    monkeypatch.delenv("STATE_DIRECTORY", raising=False)
    monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "share"))


@pytest.fixture
def store() -> Store:
    with Store(":memory:") as opened:
        yield opened


@pytest.fixture
def transport() -> FakeTransport:
    return FakeTransport()


@pytest.fixture
def clock() -> dt.datetime:
    return dt.datetime(2026, 8, 29, 12, 0, tzinfo=dt.UTC)


# The exotic payload shapes, one per key, so the builder below stays inside
# the argument limit instead of growing a boolean per frame variant.
EXTRAS: dict[str, dict[str, Any]] = {
    "sticker": {"sticker": {"packId": "abc", "stickerId": 6}},
    "reaction": {"reaction": {"emoji": "👍", "targetAuthor": "x"}},
    "quote": {"quote": {"id": 1719177917000, "author": "x", "text": "..."}},
}


def envelope(
    *,
    text: str | None = "hello",
    group_id: str | None = WATCHED_GROUP,
    sender: tuple[str, str] = ("uuid-alice", "Alice"),
    sync: bool = False,
    attachments: tuple[str, ...] = (),
    extra: str = "",
    timestamp: int = 1719177917717,
) -> str:
    """Build one raw websocket frame in either envelope shape."""
    uuid, name = sender
    payload: dict[str, Any] = {"timestamp": timestamp}
    if text is not None:
        payload["message"] = text
    if group_id is not None:
        payload["groupInfo"] = {"groupId": group_id, "type": "DELIVER"}
    if extra:
        payload.update(EXTRAS[extra])
    if attachments:
        payload["attachments"] = [
            {"contentType": c, "id": str(i)} for i, c in enumerate(attachments)
        ]

    body: dict[str, Any] = {
        "source": name,
        "sourceNumber": STRANGER_NUMBER,
        "sourceUuid": uuid,
        "sourceName": name,
        "sourceDevice": 69,
        "timestamp": timestamp,
    }
    if sync:
        body["syncMessage"] = {"sentMessage": payload}
    else:
        body["dataMessage"] = payload
    return json.dumps({"envelope": body, "account": "+15550000000"})


def update_envelope(
    *,
    group_id: str = WATCHED_GROUP,
    number: str = TRUSTED_NUMBER,
    uuid: str = "uuid-operator",
) -> str:
    """Build a group-membership-change frame: no text, groupInfo type UPDATE.

    This is the shape signal-cli emits when somebody adds the bot to a
    group, and the only frame that names who did it.
    """
    return json.dumps(
        {
            "envelope": {
                "source": number,
                "sourceNumber": number,
                "sourceUuid": uuid,
                "sourceName": "kuhy",
                "timestamp": 1719177917717,
                "dataMessage": {
                    "timestamp": 1719177917717,
                    "message": None,
                    "groupInfo": {"groupId": group_id, "type": "UPDATE"},
                },
            },
            "account": "+15550000000",
        }
    )