signal-bot/fake_signal/scenario.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Scenario files: what to inject, and what the bot must do in response."""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any

import yaml

if TYPE_CHECKING:
    from pathlib import Path


class ScenarioError(ValueError):
    """Raised when a scenario file is malformed."""


@dataclass(frozen=True, slots=True)
class Injection:
    """One inbound message to push down the websocket."""

    text: str | None = None
    uuid: str = "uuid-alice"
    name: str = "Alice"
    group: str | None = "watched"
    sticker: bool = False
    reaction: bool = False
    sync: bool = False

    def to_frame(self, groups: dict[str, str]) -> str:
        """Render this injection as a raw signal-cli envelope."""
        payload: dict[str, Any] = {"timestamp": 1719177917717}
        if self.text is not None:
            payload["message"] = self.text
        if self.group is not None:
            if self.group not in groups:
                msg = f"unknown group alias {self.group!r}; known: {sorted(groups)}"
                raise ScenarioError(msg)
            payload["groupInfo"] = {"groupId": groups[self.group], "type": "DELIVER"}
        if self.sticker:
            payload["sticker"] = {"packId": "abc", "stickerId": 6}
        if self.reaction:
            payload["reaction"] = {"emoji": "\U0001f44d", "targetAuthor": "x"}

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


@dataclass(frozen=True, slots=True)
class GroupArrival:
    """A group that appears on the account partway through a scenario.

    This is what makes "somebody added the bot to a group" expressible
    offline. ``id`` and ``internal_id`` are scenario alias names, resolved
    through the same table the injections use, so a scenario never contains
    a real group id.
    """

    id: str
    internal_id: str
    name: str = "added-later"
    member: bool = True
    joinable: bool = True

    def to_entry(self, groups: dict[str, str]) -> dict[str, Any]:
        """Render this as a /v1/groups row.

        Args:
            groups: Scenario alias to real group id.

        Returns:
            The listing row the fake server should start returning.

        Raises:
            ScenarioError: An alias is not in the table.
        """
        for alias in (self.id, self.internal_id):
            if alias not in groups:
                msg = f"unknown group alias {alias!r}; known: {sorted(groups)}"
                raise ScenarioError(msg)
        return {
            "id": groups[self.id],
            "internal_id": groups[self.internal_id],
            "name": self.name,
            "member": self.member,
            "joinable": self.joinable,
        }


@dataclass(frozen=True, slots=True)
class Expectation:
    """One message the bot is expected to have sent."""

    kind: str = "text"
    to: str = "reply"
    contains: str | None = None


@dataclass(frozen=True, slots=True)
class Scenario:
    """A named list of injections and the sends they must produce."""

    name: str
    arrive: list[GroupArrival] = field(default_factory=list)
    inject: list[Injection] = field(default_factory=list)
    expect: list[Expectation] = field(default_factory=list)


def _build[T](cls: type[T], entry: object, what: str) -> T:
    """Instantiate a scenario dataclass, rejecting unknown keys."""
    if not isinstance(entry, dict):
        msg = f"each {what} entry must be a mapping, got {type(entry).__name__}"
        raise ScenarioError(msg)
    allowed = set(cls.__annotations__)
    unknown = set(entry) - allowed
    if unknown:
        msg = f"unknown {what} keys {sorted(unknown)}; allowed: {sorted(allowed)}"
        raise ScenarioError(msg)
    return cls(**entry)


def load_scenario(path: Path) -> Scenario:
    """Load and validate one scenario file.

    Raises:
        ScenarioError: The file is not a mapping or carries unknown keys.
    """
    document = yaml.safe_load(path.read_text(encoding="utf-8"))
    if not isinstance(document, dict):
        msg = f"{path} must contain a mapping"
        raise ScenarioError(msg)
    if "name" not in document:
        msg = f"{path} is missing a name"
        raise ScenarioError(msg)
    return Scenario(
        name=str(document["name"]),
        arrive=[_build(GroupArrival, e, "arrive") for e in document.get("arrive", [])],
        inject=[_build(Injection, e, "inject") for e in document.get("inject", [])],
        expect=[_build(Expectation, e, "expect") for e in document.get("expect", [])],
    )


def describe(payload: dict[str, Any]) -> tuple[str, str, str]:
    """Reduce a captured send to (kind, recipient, body) for comparison."""
    kind = "image" if payload.get("base64_attachments") else "text"
    recipients = payload.get("recipients") or [""]
    return kind, str(recipients[0]), str(payload.get("message", ""))


def check(scenario: Scenario, sent: list[dict[str, Any]], groups: dict[str, str]) -> list[str]:
    """Compare captured sends against expectations, returning failure lines."""
    failures: list[str] = []
    if len(sent) != len(scenario.expect):
        failures.append(f"expected {len(scenario.expect)} send(s), captured {len(sent)}")
    for index, expected in enumerate(scenario.expect):
        if index >= len(sent):
            failures.append(f"[{index}] missing: expected a {expected.kind}")
            continue
        kind, recipient, body = describe(sent[index])
        want_to = groups.get(expected.to, expected.to)
        if kind != expected.kind:
            failures.append(f"[{index}] kind: expected {expected.kind}, got {kind}")
        if recipient != want_to:
            failures.append(f"[{index}] recipient: expected {want_to}, got {recipient}")
        if expected.contains is not None and expected.contains not in body:
            failures.append(f"[{index}] body: {expected.contains!r} not in {body!r}")
    return failures