mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 18:17:02 +01:00
Krzysztof kuhy Rudnicki
e3c7bbfe23
A compliment gets a reaction drawn from a small warm set (❤️ 🥰 😊 👍 🎉 🙏 😸) put on the compliment itself, and nothing in the group -- the one answer that adds no line. Same cooldown as every other command. The transport learns reactions: send_reaction on the Protocol, POST /v1/reactions/<number> on the REST adapter with the four fields api.SendReactionRequest requires (204 back), a recorded list on the fake. CommandContext.react() names the target the way a quote does, and falls back to the emoji as text for an envelope with no timestamp, so a shape this bot has never seen is visible rather than swallowed. The fake container gains the endpoint, strict about the schema, and records reactions in order with the sends; a scenario can now expect {kind: reaction, quotes: true}. First two-word trigger: fold() keeps the inner space and the registry matches whole text first, so "!good bot" resolves and "!good" does not; one-word spellings cover a double space. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""In-process transport double used by the unit tests.
|
|
|
|
This is the fast path: no sockets, no server, no event loop scheduling games.
|
|
The protocol-faithful HTTP/WebSocket fake lives in the ``fake_signal`` package
|
|
and is what the end-to-end script drives.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator, Iterable
|
|
|
|
from signal_bot.quoting import Quote
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SentMessage:
|
|
"""One outbound call captured by the fake."""
|
|
|
|
recipient: str
|
|
text: str | None = None
|
|
image_base64: str | None = None
|
|
caption: str = ""
|
|
quote: Quote | None = None
|
|
media_type: str = ""
|
|
filename: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Reaction:
|
|
"""One emoji the bot put on somebody's message."""
|
|
|
|
emoji: str
|
|
recipient: str
|
|
target_author: str
|
|
timestamp: int
|
|
|
|
|
|
@dataclass
|
|
class FakeTransport:
|
|
"""Replays canned frames and records everything the bot sends."""
|
|
|
|
frames: list[str] = field(default_factory=list)
|
|
sent: list[SentMessage] = field(default_factory=list)
|
|
# Kept apart from ``sent``: a reaction adds no message to the group, and
|
|
# a test asserting "the bot said nothing" must stay true after one.
|
|
reactions: list[Reaction] = field(default_factory=list)
|
|
attachments: list[str] = field(default_factory=list)
|
|
deleted: list[str] = field(default_factory=list)
|
|
# Attachment id to its bytes, so a test can post a real upload. An id
|
|
# that is not in here raises, which is how the "the container lost it"
|
|
# branch of !add_photo is reached.
|
|
payloads: dict[str, bytes] = field(default_factory=dict)
|
|
|
|
def queue(self, frames: Iterable[str]) -> None:
|
|
"""Add frames to be yielded by the next receive() iteration."""
|
|
self.frames.extend(frames)
|
|
|
|
async def receive(self) -> AsyncIterator[str]:
|
|
"""Yield every queued frame once, then stop."""
|
|
for frame in list(self.frames):
|
|
yield frame
|
|
|
|
async def send_text(self, text: str, recipient: str, quote: Quote | None = None) -> None:
|
|
"""Record a text send."""
|
|
self.sent.append(SentMessage(recipient=recipient, text=text, quote=quote))
|
|
|
|
async def send_image(
|
|
self,
|
|
image_base64: str,
|
|
recipient: str,
|
|
quote: Quote | None = None,
|
|
caption: str = "",
|
|
) -> None:
|
|
"""Record an image send."""
|
|
await self.send_media(image_base64, recipient, quote, caption)
|
|
|
|
async def send_media(
|
|
self,
|
|
payload_base64: str,
|
|
recipient: str,
|
|
quote: Quote | None = None,
|
|
caption: str = "",
|
|
media_type: str = "",
|
|
filename: str = "",
|
|
) -> None:
|
|
"""Record an attachment send, type and filename included."""
|
|
self.sent.append(
|
|
SentMessage(
|
|
recipient=recipient,
|
|
image_base64=payload_base64,
|
|
quote=quote,
|
|
caption=caption,
|
|
media_type=media_type,
|
|
filename=filename,
|
|
)
|
|
)
|
|
|
|
async def send_reaction(
|
|
self, emoji: str, recipient: str, target_author: str, timestamp: int
|
|
) -> None:
|
|
"""Record a reaction."""
|
|
self.reactions.append(Reaction(emoji, recipient, target_author, timestamp))
|
|
|
|
async def list_attachments(self) -> list[str]:
|
|
"""Return the configured attachment ids."""
|
|
return list(self.attachments)
|
|
|
|
async def fetch_attachment(self, attachment_id: str) -> bytes:
|
|
"""Return one stored attachment's bytes.
|
|
|
|
Raises:
|
|
KeyError: Nothing was stored under that id.
|
|
"""
|
|
return self.payloads[attachment_id]
|
|
|
|
async def delete_attachment(self, attachment_id: str) -> None:
|
|
"""Record an attachment deletion."""
|
|
self.deleted.append(attachment_id)
|
|
self.payloads.pop(attachment_id, None)
|
|
|
|
@property
|
|
def texts(self) -> list[str]:
|
|
"""Every text body sent so far, in order."""
|
|
return [m.text for m in self.sent if m.text is not None]
|