feat: !cisza silences the bot in one group for up to ten minutes
Some checks are pending
CI / check (push) Waiting to run
CI / web (push) Waiting to run

Anybody may ask; ten quiet minutes is the whole downside, so there is no
trust gate on the way in and no cooldown. The confirmation is the last
thing said: below it the dispatcher drops every reply in that group --
commands, the near-miss suggestion, and the nightly report, which is
held back (never marked) rather than lost. Messages are still counted.

!odcisz is the one command that runs through a mute, and only for
TRUSTED_INVITERS; a stranger is refused without a reply, because a
refusal is a reply. Unset, anybody may -- the same reading InviteTrust
gives an unset variable, now shared as trust.is_trusted.

There is deliberately no !unmute alias: it is one edit from !mute, and
tests/test_nearest.py forbids two commands within reach of each other.

The state is a deadline in group_meta, so it survives a restart and a
reload and is cleared with the group. tests/test_help_command.py hit
the 250-line cap and its album half moved to tests/test_help_album.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-09-04 10:31:08 +02:00
parent 0717df646a
commit 41df308393
16 changed files with 758 additions and 99 deletions

View File

@ -32,6 +32,8 @@ This file is the trigger reference — every spelling that resolves, which
| help | `!pomoc` `!pomocy` `!help` `!komendy` `!commands` |
| random | `!losowe` `!losuj` `!random` `!rand` |
| suggest | `!zaproponuj` `!propozycja` `!sugestia` `!suggest` `!pomysl` |
| mute | `!cisza` `!cicho` `!mute` |
| unmute | `!odcisz` `!gadaj` `!wracaj` |
Aliases live in `signal_bot/aliases.json`, not in source, because they are
data: adding one should never mean touching Python. Two constraints on a new
@ -71,6 +73,16 @@ not refused and never logged. It takes arguments, and those
arguments are resolved to pseudonym keys by the registry before any handler,
log line or `command_uses` row sees them. See `DOCS-photo-pack.md`.
`mute` (`!cisza [1-10]`) silences the bot in **that group only**, for the
minutes asked or ten by default, and ten at most. It confirms once and then
answers nothing: no commands, no near-miss suggestion, and the nightly report
is held back (not marked, so the next run still says that day). Messages are
still counted. Anybody may ask -- ten quiet minutes is the whole downside.
`unmute` (`!odcisz`) is the one command that runs through a mute, and only
for `TRUSTED_INVITERS`; anyone else is refused without a reply, because a
refusal is a reply. With `TRUSTED_INVITERS` unset, anybody may. There is no
`!unmute` alias: it sits one edit from `!mute`, inside the near-miss budget.
`add_photo` (`!dodaj`) is **unwired since 2026-09-04**: a photo posted to a
group must not reach the album without somebody approving it, and no approval
queue exists yet. It comes back when one does; `DOCS-add-photo.md` describes

View File

@ -0,0 +1,16 @@
# One confirmation, then nothing -- not even for a command that would
# otherwise be served -- until the unmute, which is the only word that gets
# through. The suite runs TRUSTED_INVITERS empty, so anybody may end it,
# and the scenario MUST end unmuted: every later file shares this process
# and this group, and a mute left standing would silence all of them.
name: "!cisza silences the group until !odcisz"
inject:
- { text: "!cisza 1", uuid: uuid-mona, name: Mona }
- { text: "!cat", uuid: uuid-mona, name: Mona }
- { text: "!cta", uuid: uuid-mona, name: Mona }
- { text: "!odcisz", uuid: uuid-mona, name: Mona }
- { text: "!cat", uuid: uuid-mona, name: Mona }
expect:
- { kind: text, to: reply, contains: "CISZA PRZEZ 1 MINUTĘ" }
- { kind: text, to: reply, contains: "WRACAM" }
- { kind: image, to: reply }

View File

@ -157,5 +157,15 @@
"!sugestia",
"!suggest",
"!pomysl"
],
"mute": [
"!cisza",
"!cicho",
"!mute"
],
"unmute": [
"!odcisz",
"!gadaj",
"!wracaj"
]
}

View File

@ -18,6 +18,7 @@ from signal_bot.announce import announce_all
from signal_bot.changelog import load as load_changelog
from signal_bot.dispatcher import Dispatcher
from signal_bot.http_client import ImageFetcher
from signal_bot.mute import is_muted
from signal_bot.portraits import PortraitCache
from signal_bot.registry_holder import RegistryHolder
from signal_bot.reporting import compose
@ -104,6 +105,12 @@ class Bot:
if not days:
return
now = self.clock()
# Deferred, not dropped: nothing is marked, so the next run says
# these days along with its own. A mute is ten minutes at most, and
# the report is the one message a mute must be able to hold back.
if is_muted(self.store, group_id, now.timestamp()):
logger.info("report for %s held back: the group is muted", group_id)
return
text = compose(
self.store,
self.directory,

View File

@ -9,6 +9,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
from signal_bot import mute
from signal_bot.analytics import probe, recorder
from signal_bot.context import CommandContext
from signal_bot.counting import should_count, watched_group
@ -128,6 +129,14 @@ class Dispatcher:
is_command=command is not None,
)
# Below the recording, above every reply: a muted group is still
# listened to and counted, it is just not talked to. The one command
# that passes is the one that ends the mute.
now = self.clock()
if not mute.allows(self.store, group, now.timestamp(), command):
logger.debug("muted in %s; not answering", group)
return
if command is None:
# is_command above stays False, and this stays below the
# recording rather than in front of it: an unrecognised word is
@ -137,7 +146,6 @@ class Dispatcher:
return
logger.info("running command %s for %s", command.name, message.source_name)
now = self.clock()
ctx = CommandContext(
message=message,
group_id=group,

View File

@ -42,6 +42,7 @@ GROUPS: Final[tuple[tuple[str, tuple[str, ...]], ...]] = (
("Papież", ("pope",)),
("Wiersze", ("poem",)),
("Propozycje", ("suggest",)),
("Cisza", ("mute", "unmute")),
(
"Statystyki",
(

69
signal_bot/mute.py Normal file
View File

@ -0,0 +1,69 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Per-group silence: the state, and the one question the dispatcher asks it.
A mute is a deadline in ``group_meta``, keyed by the canonical group id, so
it survives a restart and a reload alike and never leaks into another
group. While it stands the bot says nothing in that group: no replies, no
"did you mean", no nightly report. Messages are still counted -- silence is
about talking, not about listening -- and exactly one command still runs,
the one that ends it.
Anyone may ask for silence, for at most ten minutes, because the worst
anyone can do with that is buy ten quiet minutes. Ending it early is the
operator's call: see mute_command.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from signal_bot.registry import Command
from signal_bot.store import Store
# The group_meta key. Epoch seconds, as text, like every other value there.
MUTED_UNTIL_KEY: Final = "muted_until"
MAX_MINUTES: Final = 10
DEFAULT_MINUTES: Final = MAX_MINUTES
# The one command that runs through a mute. By name rather than by a flag on
# Command, so that the registry -- which has nothing to do with silence --
# does not grow a field only one command sets.
UNMUTE_NAME: Final = "unmute"
def muted_until(store: Store, group_id: str) -> float | None:
"""When the current mute ends, or None when there has never been one."""
raw = store.meta(group_id, MUTED_UNTIL_KEY)
return None if raw is None else float(raw)
def is_muted(store: Store, group_id: str, now: float) -> bool:
"""Whether the group is silent at this instant.
An expired deadline is left in place rather than deleted: reading it
costs one comparison, and deleting it would make this a write.
"""
until = muted_until(store, group_id)
return until is not None and now < until
def mute(store: Store, group_id: str, until: float) -> None:
"""Silence the group until the given instant, replacing any earlier deadline."""
store.set_meta(group_id, MUTED_UNTIL_KEY, repr(until))
def unmute(store: Store, group_id: str) -> None:
"""End the silence now. Idempotent: a group that is not muted stays not muted."""
store.set_meta(group_id, MUTED_UNTIL_KEY, "0")
def allows(store: Store, group_id: str, now: float, command: Command | None) -> bool:
"""Whether the dispatcher may say anything in reply to this message.
True whenever the group is not muted; while it is, true only for the
command that ends the mute. ``None`` is a word nobody registered, whose
"did you mean" reply is exactly the kind of noise a mute is for.
"""
if not is_muted(store, group_id, now):
return True
return command is not None and command.name == UNMUTE_NAME

105
signal_bot/mute_command.py Normal file
View File

@ -0,0 +1,105 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""!cisza and !odcisz: ask the bot to be quiet for a while, and let it speak again.
Two commands with different owners. Anybody can mute, because a bot that
posts a photo into a conversation that has turned serious needs a way to be
told "not now" by whoever is in it, and a ten-minute cap is the whole
safeguard -- there is nothing to abuse in ten quiet minutes. Ending a mute
early belongs to the operator (TRUSTED_INVITERS), because otherwise the
mute is only as long as the next person's patience.
The mute confirms once and then the bot says nothing at all in that group
until the deadline: the dispatcher drops every reply, including "did you
mean" and the nightly report (see mute.allows and Bot._report_group). A
release announcement is the one thing not covered: it is sent at startup
and never repeated, so skipping it would lose it rather than delay it.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Final
from signal_bot.animals import load_aliases
from signal_bot.mute import DEFAULT_MINUTES, MAX_MINUTES, UNMUTE_NAME, is_muted, mute, unmute
from signal_bot.stats.text import plural
from signal_bot.trust import is_trusted
if TYPE_CHECKING:
from signal_bot.context import CommandContext
from signal_bot.registry import CommandRegistry
logger = logging.getLogger(__name__)
MUTE_NAME: Final = "mute"
MUTE_DESCRIPTION: Final = f"ucisza bota na N minut (najwyżej {MAX_MINUTES}), np. !cisza 5"
UNMUTE_DESCRIPTION: Final = "kończy ciszę przed czasem (tylko zaufani)"
MUTED: Final = "BEEP BOOP CISZA PRZEZ {minutes} {unit}."
USAGE: Final = f"BEEP BOOP !cisza <1-{MAX_MINUTES}>."
NOT_MUTED: Final = "BEEP BOOP NIE JESTEM UCISZONY."
UNMUTED: Final = "BEEP BOOP WRACAM."
def minutes_from(args: tuple[str, ...]) -> int | None:
"""How long was asked for, clamped to the cap; None when it was not a number.
No argument is the full ten minutes: somebody typing the bare word wants
quiet, and asking them to be precise about it defeats the point.
"""
if not args:
return DEFAULT_MINUTES
try:
asked = int(args[0])
except ValueError:
return None
if asked < 1:
return None
return min(asked, MAX_MINUTES)
async def handle_mute(ctx: CommandContext) -> None:
"""Silence the group for the requested time, and say so exactly once."""
minutes = minutes_from(ctx.args)
if minutes is None:
await ctx.reply(USAGE)
return
mute(ctx.store, ctx.group_id, ctx.now + minutes * 60)
logger.info("%s muted %s for %d min", ctx.message.source_name, ctx.group_id, minutes)
await ctx.reply(
MUTED.format(minutes=minutes, unit=plural(minutes, "MINUTĘ", "MINUTY", "MINUT"))
)
async def handle_unmute(ctx: CommandContext) -> None:
"""End the silence early -- for the operator; in silence for anybody else.
A refused unmute gets no reply on purpose: the group is muted, and
"you may not unmute me" is exactly the reply a mute forbids. It is
logged instead. An unset TRUSTED_INVITERS means anybody may, the same
reading InviteTrust gives an unset variable.
"""
if not is_muted(ctx.store, ctx.group_id, ctx.now):
await ctx.reply(NOT_MUTED)
return
trusted = ctx.config.trusted_inviters
message = ctx.message
if trusted and not is_trusted(trusted, message.source_number, message.source_uuid):
logger.info("%s asked to unmute %s and is not trusted", message.source_name, ctx.group_id)
return
unmute(ctx.store, ctx.group_id)
logger.info("%s unmuted %s", message.source_name, ctx.group_id)
await ctx.reply(UNMUTED)
def register_mute_commands(
registry: CommandRegistry, *, aliases: dict[str, list[str]] | None = None
) -> None:
"""Register both commands, deliberately outside the cooldown.
A cooldown would mean "wait ten seconds before you may ask for quiet",
and the unmute is gated by trust rather than by time.
"""
table = load_aliases() if aliases is None else aliases
registry.register(MUTE_NAME, table[MUTE_NAME], handle_mute, MUTE_DESCRIPTION, takes_args=True)
registry.register(UNMUTE_NAME, table[UNMUTE_NAME], handle_unmute, UNMUTE_DESCRIPTION)

View File

@ -29,10 +29,13 @@ from __future__ import annotations
import logging
import re
from typing import Any, Final
from typing import TYPE_CHECKING, Any, Final
from signal_bot.groups import ConfigError, canonical_group_id
if TYPE_CHECKING:
from collections.abc import Collection
logger = logging.getLogger(__name__)
TRUSTED_INVITERS_KEY: Final = "TRUSTED_INVITERS"
@ -74,6 +77,16 @@ def parse_trusted_inviters(env: dict[str, str]) -> tuple[str, ...]:
return tuple(trusted)
def is_trusted(trusted: Collection[str], number: str, uuid: str) -> bool:
"""Whether a sender is on the TRUSTED_INVITERS list, by either identity.
Strict: an empty list trusts nobody here. What an *unset* variable means
is the caller's policy -- InviteTrust reads it as "every group", the
unmute command as "anybody" -- and each says so where it decides.
"""
return number in trusted or uuid in trusted
def identities(entry: dict[str, Any], key: str) -> set[str]:
"""Return one listing field as a set of strings, tolerating anything.
@ -117,7 +130,7 @@ class InviteTrust:
source_number: The +E.164 number that made the change, if known.
source_uuid: The uuid that made the change.
"""
if source_number not in self._trusted and source_uuid not in self._trusted:
if not is_trusted(self._trusted, source_number, source_uuid):
return
canonical = canonical_group_id(group_id)
if canonical not in self._vouched:

View File

@ -18,6 +18,7 @@ from signal_bot.animals import load_aliases, register_animal_commands
from signal_bot.art_command import register_art_command
from signal_bot.decorators import cooldown
from signal_bot.help_command import register_help_command
from signal_bot.mute_command import register_mute_commands
from signal_bot.pack_command import register_pack_commands
from signal_bot.pack_holder import PackHolder
from signal_bot.pack_load import load as load_pack
@ -84,6 +85,9 @@ def build_registry(
# row is asked to wait for the second. The alternative is one person
# being able to fill a table nothing prunes.
register_suggest_command(registry, decorate=rate_limit)
# No cooldown: "wait before you may ask for quiet" is the wrong answer,
# and the unmute is gated by trust, not by time.
register_mute_commands(registry)
# Absent unless this installation has an album, and scoped to that
# album's own groups when it does -- so on every other machine, and in
# every other group, these commands simply do not exist.

View File

@ -12,6 +12,7 @@ import httpx
from signal_bot.bot import Bot, build_bot
from signal_bot.discovery import GroupDirectory
from signal_bot.liveness import Liveness
from signal_bot.mute import mute, unmute
from signal_bot.stats.events import MessageEvent
from signal_bot.stats.nightly import LAST_REPORTED_KEY
from signal_bot.stats.periods import WARSAW
@ -175,3 +176,22 @@ async def test_the_report_follows_the_live_group_set(
wrote(store, SECOND_GROUP, "u2", "Bob", 1)
await _bot(config, store, transport, directory).publish_report()
assert [m.recipient for m in transport.sent] == [WATCHED_GROUP, SECOND_GROUP]
async def test_a_muted_group_has_its_report_held_back_not_lost(
config: Config, store: Store, transport: FakeTransport, directory: GroupDirectory
) -> None:
"""Nothing is sent and nothing is marked, so the next run still owes the day."""
wrote(store, WATCHED_GROUP, "u1", "Alice", 1)
wrote(store, SECOND_GROUP, "u2", "Bob", 1)
mute(store, WATCHED_GROUP, NOW.timestamp() + 60)
bot = _bot(config, store, transport, directory)
await bot.publish_report()
assert [m.recipient for m in transport.sent] == [SECOND_GROUP]
assert store.meta(WATCHED_GROUP, LAST_REPORTED_KEY) is None
unmute(store, WATCHED_GROUP)
await bot.publish_report()
assert [m.recipient for m in transport.sent] == [SECOND_GROUP, WATCHED_GROUP]
assert "Alice" in transport.texts[1]

View File

@ -0,0 +1,131 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The dispatcher while a group is muted: listening, counting, not talking."""
from __future__ import annotations
import datetime as dt
from typing import TYPE_CHECKING
from signal_bot.dispatcher import Dispatcher
from signal_bot.mute import mute
from signal_bot.mute_command import register_mute_commands
from signal_bot.registry import CommandRegistry
from signal_bot.registry_holder import RegistryHolder
from signal_bot.stats.queries_day import day_rows
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, envelope
if TYPE_CHECKING:
from signal_bot.config import Config
from signal_bot.context import CommandContext
from signal_bot.discovery import GroupDirectory
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
NOW = dt.datetime(2026, 9, 4, 12, 0, tzinfo=dt.UTC)
ALIASES = {"mute": ["!cisza"], "unmute": ["!odcisz"]}
def _dispatcher(
config: Config,
directory: GroupDirectory,
store: Store,
transport: FakeTransport,
runs: list[str],
clock: list[dt.datetime],
) -> Dispatcher:
"""!cat plus the mute pair, in front of the real dispatcher."""
async def handler(ctx: CommandContext) -> None:
runs.append(ctx.command_name)
registry = CommandRegistry()
registry.register("cat", ["!cat"], handler, "kot")
register_mute_commands(registry, aliases=ALIASES)
return Dispatcher(
config=config,
directory=directory,
holder=RegistryHolder(lambda: registry),
store=store,
transport=transport,
clock=lambda: clock[0],
)
async def test_a_muted_group_is_counted_but_not_answered(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
"""A command, an unknown word, chatter: three messages recorded, no reply."""
runs: list[str] = []
dispatcher = _dispatcher(config, directory, store, transport, runs, [NOW])
mute(store, WATCHED_GROUP, NOW.timestamp() + 60)
await dispatcher.handle(envelope(text="!cat", timestamp=1))
await dispatcher.handle(envelope(text="!cta", timestamp=2))
await dispatcher.handle(envelope(text="hello", timestamp=3))
assert runs == []
assert transport.sent == []
day = store.events.first_day(WATCHED_GROUP)
assert day is not None
(row,) = day_rows(store.events, WATCHED_GROUP, day)
assert (row.messages, row.commands) == (3, 1)
# The near miss is not filed either: nothing about the mute is remembered
# as a suggestion to act on later.
assert store.suggestions.unknown() == []
async def test_the_mute_is_confirmed_once_and_then_the_bot_is_silent(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
runs: list[str] = []
dispatcher = _dispatcher(config, directory, store, transport, runs, [NOW])
await dispatcher.handle(envelope(text="!cisza 1"))
await dispatcher.handle(envelope(text="!cisza 1"))
await dispatcher.handle(envelope(text="!cat"))
assert [m.text for m in transport.sent] == ["BEEP BOOP CISZA PRZEZ 1 MINUTĘ."]
assert runs == []
async def test_unmute_is_the_one_command_that_runs_through_a_mute(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
runs: list[str] = []
dispatcher = _dispatcher(config, directory, store, transport, runs, [NOW])
mute(store, WATCHED_GROUP, NOW.timestamp() + 600)
await dispatcher.handle(envelope(text="!odcisz"))
await dispatcher.handle(envelope(text="!cat"))
assert transport.sent[0].text == "BEEP BOOP WRACAM."
assert runs == ["cat"]
async def test_the_silence_ends_on_its_own_at_the_deadline(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
runs: list[str] = []
clock = [NOW]
dispatcher = _dispatcher(config, directory, store, transport, runs, clock)
await dispatcher.handle(envelope(text="!cisza 1"))
clock[0] = NOW + dt.timedelta(seconds=59)
await dispatcher.handle(envelope(text="!cat"))
assert runs == []
clock[0] = NOW + dt.timedelta(seconds=60)
await dispatcher.handle(envelope(text="!cat"))
assert runs == ["cat"]
async def test_a_mute_in_one_group_leaves_the_other_talking(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
runs: list[str] = []
dispatcher = _dispatcher(config, directory, store, transport, runs, [NOW])
await dispatcher.handle(envelope(text="!cisza", group_id=WATCHED_GROUP))
await dispatcher.handle(envelope(text="!cat", group_id=WATCHED_GROUP))
await dispatcher.handle(envelope(text="!cat", group_id=SECOND_GROUP))
assert runs == ["cat"]

109
tests/test_help_album.py Normal file
View File

@ -0,0 +1,109 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""!pomoc with a photo album: its own headings, and invisible elsewhere.
Split out of test_help_command.py at the 250-line cap; the seam is "what
the album adds to the list", which the other file never registers.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot.help_command import ALBUM_HEADING, PACK_HEADING, render_help
from signal_bot.pack_add import register_add_command
from signal_bot.pack_command import register_pack_commands
from signal_bot.pack_holder import PackHolder
from signal_bot.pack_load import load as load_pack
from signal_bot.pack_random import register_random_command
from tests import pack_fixture
from tests.conftest import SECOND_GROUP, WATCHED_GROUP
from tests.test_help_command import full_registry
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.media_sniff import MediaKind
def test_an_album_gets_its_own_heading_and_a_hint(tmp_path: Path) -> None:
"""And the hint is built from real triggers, because they are the example."""
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(pack_fixture.build(tmp_path / "pack"))))
text = render_help(registry.commands, WATCHED_GROUP)
assert f"\n{PACK_HEADING}\n" in text
assert "!alfa — losowe zdjęcie tej osoby" in text
assert "Kilka naraz (!alfa !beta)" in text
def test_the_album_is_invisible_in_every_other_group(tmp_path: Path) -> None:
"""The list is what would announce a private command family to strangers."""
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(pack_fixture.build(tmp_path / "pack"))))
text = render_help(registry.commands, SECOND_GROUP)
assert PACK_HEADING not in text
assert "alfa" not in text
assert "!kot" in text
def test_a_pack_of_one_prints_no_hint(tmp_path: Path) -> None:
""" "Combine them" needs two things to combine."""
root = pack_fixture.build(
tmp_path / "pack",
people=[{"key": "p1", "triggers": ["!alfa"]}],
combos=[],
tags={},
)
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(root)))
text = render_help(registry.commands, WATCHED_GROUP)
assert PACK_HEADING in text
assert "Kilka naraz" not in text
class _NoStripper:
"""Registration needs a stripper; rendering help never calls one."""
def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
return payload
def test_the_album_commands_get_their_own_block(tmp_path: Path) -> None:
"""Their own heading, so the "kilka naraz" hint keeps describing people.
And neither description may name anybody: !pomoc prints them, and the
trigger is the only place a name is allowed to appear.
"""
holder = PackHolder(load_pack(pack_fixture.build(tmp_path / "pack")))
registry = full_registry()
register_pack_commands(registry, holder)
register_random_command(registry, holder, ("!losowe",))
register_add_command(registry, holder, _NoStripper(), ("!dodaj",))
text = render_help(registry.commands, WATCHED_GROUP)
assert f"\n{ALBUM_HEADING}\n" in text
assert "!dodaj — dodaje załączone zdjęcie do albumu z podanymi etykietami" in text
assert "!losowe — losowe zdjęcie z albumu, opcjonalnie po etykiecie" in text
# The people's block is still its own, with its own hint.
assert f"\n{PACK_HEADING}\n" in text
assert "Kilka naraz (!alfa !beta)" in text
def test_the_album_commands_are_invisible_in_every_other_group(tmp_path: Path) -> None:
holder = PackHolder(load_pack(pack_fixture.build(tmp_path / "pack")))
registry = full_registry()
register_pack_commands(registry, holder)
register_random_command(registry, holder, ("!losowe",))
register_add_command(registry, holder, _NoStripper(), ("!dodaj",))
text = render_help(registry.commands, SECOND_GROUP)
assert ALBUM_HEADING not in text
assert "!dodaj" not in text
assert "!losowe" not in text

View File

@ -14,22 +14,16 @@ from signal_bot.art_command import register_art_command
from signal_bot.config import DEFAULT_COMMONS_API_URL, DEFAULT_POEM_TEXT_URL
from signal_bot.context import CommandContext
from signal_bot.help_command import (
ALBUM_HEADING,
COMMAND_NAME,
GROUPS,
HEADING,
PACK_HEADING,
SPELLINGS,
register_help_command,
render_help,
)
from signal_bot.http_client import ImageFetcher
from signal_bot.models import parse_envelope
from signal_bot.pack_add import register_add_command
from signal_bot.pack_command import register_pack_commands
from signal_bot.pack_holder import PackHolder
from signal_bot.pack_load import load as load_pack
from signal_bot.pack_random import register_random_command
from signal_bot.mute_command import register_mute_commands
from signal_bot.poem_command import register_poem_command
from signal_bot.popes_command import register_pope_command
from signal_bot.portraits import PortraitCache
@ -37,14 +31,10 @@ from signal_bot.registry import CommandRegistry
from signal_bot.stats.commands import register_stats_commands
from signal_bot.suggest_command import register_suggest_command
from signal_bot.traps_command import register_trap_command
from tests import pack_fixture
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, envelope
from tests.conftest import WATCHED_GROUP, envelope
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.config import Config
from signal_bot.media_sniff import MediaKind
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
@ -74,6 +64,7 @@ def full_registry() -> CommandRegistry:
register_stats_commands(registry)
register_analytics_command(registry)
register_suggest_command(registry)
register_mute_commands(registry)
register_help_command(registry)
return registry
@ -164,87 +155,3 @@ def test_every_alias_resolves(trigger: str) -> None:
command = full_registry().resolve(trigger)
assert command is not None
assert command.name == COMMAND_NAME
def test_an_album_gets_its_own_heading_and_a_hint(tmp_path: Path) -> None:
"""And the hint is built from real triggers, because they are the example."""
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(pack_fixture.build(tmp_path / "pack"))))
text = render_help(registry.commands, WATCHED_GROUP)
assert f"\n{PACK_HEADING}\n" in text
assert "!alfa — losowe zdjęcie tej osoby" in text
assert "Kilka naraz (!alfa !beta)" in text
def test_the_album_is_invisible_in_every_other_group(tmp_path: Path) -> None:
"""The list is what would announce a private command family to strangers."""
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(pack_fixture.build(tmp_path / "pack"))))
text = render_help(registry.commands, SECOND_GROUP)
assert PACK_HEADING not in text
assert "alfa" not in text
assert "!kot" in text
def test_a_pack_of_one_prints_no_hint(tmp_path: Path) -> None:
""" "Combine them" needs two things to combine."""
root = pack_fixture.build(
tmp_path / "pack",
people=[{"key": "p1", "triggers": ["!alfa"]}],
combos=[],
tags={},
)
registry = full_registry()
register_pack_commands(registry, PackHolder(load_pack(root)))
text = render_help(registry.commands, WATCHED_GROUP)
assert PACK_HEADING in text
assert "Kilka naraz" not in text
class _NoStripper:
"""Registration needs a stripper; rendering help never calls one."""
def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
return payload
def test_the_album_commands_get_their_own_block(tmp_path: Path) -> None:
"""Their own heading, so the "kilka naraz" hint keeps describing people.
And neither description may name anybody: !pomoc prints them, and the
trigger is the only place a name is allowed to appear.
"""
holder = PackHolder(load_pack(pack_fixture.build(tmp_path / "pack")))
registry = full_registry()
register_pack_commands(registry, holder)
register_random_command(registry, holder, ("!losowe",))
register_add_command(registry, holder, _NoStripper(), ("!dodaj",))
text = render_help(registry.commands, WATCHED_GROUP)
assert f"\n{ALBUM_HEADING}\n" in text
assert "!dodaj — dodaje załączone zdjęcie do albumu z podanymi etykietami" in text
assert "!losowe — losowe zdjęcie z albumu, opcjonalnie po etykiecie" in text
# The people's block is still its own, with its own hint.
assert f"\n{PACK_HEADING}\n" in text
assert "Kilka naraz (!alfa !beta)" in text
def test_the_album_commands_are_invisible_in_every_other_group(tmp_path: Path) -> None:
holder = PackHolder(load_pack(pack_fixture.build(tmp_path / "pack")))
registry = full_registry()
register_pack_commands(registry, holder)
register_random_command(registry, holder, ("!losowe",))
register_add_command(registry, holder, _NoStripper(), ("!dodaj",))
text = render_help(registry.commands, SECOND_GROUP)
assert ALBUM_HEADING not in text
assert "!dodaj" not in text
assert "!losowe" not in text

88
tests/test_mute.py Normal file
View File

@ -0,0 +1,88 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The mute state: a deadline per group, and the one question asked of it."""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot.mute import (
MUTED_UNTIL_KEY,
UNMUTE_NAME,
allows,
is_muted,
mute,
muted_until,
unmute,
)
from signal_bot.registry import CommandRegistry
from tests.conftest import SECOND_GROUP, WATCHED_GROUP
if TYPE_CHECKING:
from signal_bot.context import CommandContext
from signal_bot.store import Store
NOW = 1_000.0
async def _noop(ctx: CommandContext) -> None:
"""A handler that does nothing."""
def test_a_group_that_was_never_muted_has_no_deadline(store: Store) -> None:
assert muted_until(store, WATCHED_GROUP) is None
assert not is_muted(store, WATCHED_GROUP, NOW)
def test_a_mute_stands_until_its_deadline_and_not_a_moment_longer(store: Store) -> None:
mute(store, WATCHED_GROUP, NOW + 60)
assert muted_until(store, WATCHED_GROUP) == NOW + 60
assert is_muted(store, WATCHED_GROUP, NOW)
assert is_muted(store, WATCHED_GROUP, NOW + 59.9)
# The deadline itself is over: "until" is exclusive.
assert not is_muted(store, WATCHED_GROUP, NOW + 60)
def test_a_mute_is_per_group(store: Store) -> None:
"""Silence asked for in one group must not reach into another."""
mute(store, WATCHED_GROUP, NOW + 60)
assert is_muted(store, WATCHED_GROUP, NOW)
assert not is_muted(store, SECOND_GROUP, NOW)
def test_a_later_mute_replaces_the_deadline_rather_than_extending_it(store: Store) -> None:
"""The cap is a cap: ten minutes asked twice is still ten minutes from now."""
mute(store, WATCHED_GROUP, NOW + 600)
mute(store, WATCHED_GROUP, NOW + 60)
assert muted_until(store, WATCHED_GROUP) == NOW + 60
def test_unmute_ends_the_silence_now_and_is_idempotent(store: Store) -> None:
mute(store, WATCHED_GROUP, NOW + 600)
unmute(store, WATCHED_GROUP)
assert not is_muted(store, WATCHED_GROUP, NOW)
unmute(store, WATCHED_GROUP)
assert not is_muted(store, WATCHED_GROUP, NOW)
def test_the_deadline_lives_in_group_meta_as_epoch_seconds(store: Store) -> None:
"""Pinned because forget_group clears group_meta, which is the intended cleanup."""
mute(store, WATCHED_GROUP, NOW + 60)
assert store.meta(WATCHED_GROUP, MUTED_UNTIL_KEY) == repr(NOW + 60)
def test_everything_is_allowed_while_nobody_asked_for_silence(store: Store) -> None:
registry = CommandRegistry()
registry.register("cat", ["!kot"], _noop, "kot")
assert allows(store, WATCHED_GROUP, NOW, registry.resolve("!kot"))
assert allows(store, WATCHED_GROUP, NOW, None)
def test_only_the_unmute_command_is_allowed_through_a_mute(store: Store) -> None:
"""A command, an unknown word and the unmute: one of the three passes."""
registry = CommandRegistry()
registry.register("cat", ["!kot"], _noop, "kot")
registry.register(UNMUTE_NAME, ["!odcisz"], _noop, "odcisz")
mute(store, WATCHED_GROUP, NOW + 60)
assert not allows(store, WATCHED_GROUP, NOW, registry.resolve("!kot"))
assert not allows(store, WATCHED_GROUP, NOW, None)
assert allows(store, WATCHED_GROUP, NOW, registry.resolve("!odcisz"))

159
tests/test_mute_command.py Normal file
View File

@ -0,0 +1,159 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""!cisza and !odcisz: who may ask for quiet, for how long, and who may end it."""
from __future__ import annotations
from dataclasses import replace
from typing import TYPE_CHECKING
import pytest
from signal_bot.context import CommandContext
from signal_bot.models import parse_envelope
from signal_bot.mute import MAX_MINUTES, is_muted, mute, muted_until
from signal_bot.mute_command import (
MUTE_NAME,
NOT_MUTED,
UNMUTED,
USAGE,
minutes_from,
register_mute_commands,
)
from signal_bot.registry import CommandRegistry
from tests.conftest import STRANGER_NUMBER, TRUSTED_NUMBER, WATCHED_GROUP, envelope
if TYPE_CHECKING:
from signal_bot.config import Config
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
NOW = 1_000.0
ALIASES = {MUTE_NAME: ["!cisza"], "unmute": ["!odcisz"]}
def _registry() -> CommandRegistry:
registry = CommandRegistry()
register_mute_commands(registry, aliases=ALIASES)
return registry
async def _run(
store: Store, transport: FakeTransport, config: Config, text: str, *, number: str = ""
) -> str | None:
"""Type one line at the bot and return what it answered, or None for silence."""
message = parse_envelope(envelope(text=text, timestamp=77))
assert message is not None
if number:
message = replace(message, source_number=number)
resolved = _registry().resolve_with_args(text, WATCHED_GROUP)
assert resolved is not None
before = len(transport.sent)
ctx = CommandContext(
message=message,
group_id=WATCHED_GROUP,
transport=transport,
store=store,
config=config,
command_name=resolved.command.name,
now=NOW,
day="2026-09-04",
args=resolved.args,
)
await resolved.command.handler(ctx)
return transport.sent[-1].text if len(transport.sent) > before else None
@pytest.mark.parametrize(
("args", "minutes"),
[
((), MAX_MINUTES),
(("5",), 5),
(("1",), 1),
(("10",), MAX_MINUTES),
(("11",), MAX_MINUTES),
(("9999",), MAX_MINUTES),
(("0",), None),
(("-3",), None),
(("chwile",), None),
(("5", "minut"), 5),
],
)
def test_minutes_asked_for_are_clamped_to_the_cap(
args: tuple[str, ...], minutes: int | None
) -> None:
assert minutes_from(args) == minutes
async def test_the_bare_word_buys_the_full_ten_minutes(
store: Store, transport: FakeTransport, config: Config
) -> None:
answer = await _run(store, transport, config, "!cisza")
assert answer == "BEEP BOOP CISZA PRZEZ 10 MINUT."
assert muted_until(store, WATCHED_GROUP) == NOW + 10 * 60
@pytest.mark.parametrize(
("text", "answer", "seconds"),
[
("!cisza 1", "BEEP BOOP CISZA PRZEZ 1 MINUTĘ.", 60),
("!cisza 3", "BEEP BOOP CISZA PRZEZ 3 MINUTY.", 180),
("!cisza 7", "BEEP BOOP CISZA PRZEZ 7 MINUT.", 420),
("!cisza 60", "BEEP BOOP CISZA PRZEZ 10 MINUT.", 600),
],
)
async def test_the_confirmation_names_the_minutes_actually_granted(
store: Store, transport: FakeTransport, config: Config, text: str, answer: str, seconds: int
) -> None:
"""The one message before the silence must say how long it is, in Polish."""
assert await _run(store, transport, config, text) == answer
assert muted_until(store, WATCHED_GROUP) == NOW + seconds
async def test_a_number_that_is_not_a_number_gets_the_usage_line(
store: Store, transport: FakeTransport, config: Config
) -> None:
assert await _run(store, transport, config, "!cisza chwile") == USAGE
assert not is_muted(store, WATCHED_GROUP, NOW)
async def test_unmuting_a_group_that_is_not_muted_says_so(
store: Store, transport: FakeTransport, config: Config
) -> None:
assert await _run(store, transport, config, "!odcisz") == NOT_MUTED
async def test_anybody_may_unmute_when_nobody_is_trusted(
store: Store, transport: FakeTransport, config: Config
) -> None:
"""An unset TRUSTED_INVITERS is the gate being off, as it is for invitations."""
mute(store, WATCHED_GROUP, NOW + 600)
assert await _run(store, transport, config, "!odcisz", number=STRANGER_NUMBER) == UNMUTED
assert not is_muted(store, WATCHED_GROUP, NOW)
async def test_a_trusted_operator_ends_the_silence_early(
store: Store, transport: FakeTransport, config: Config
) -> None:
trusting = replace(config, trusted_inviters=(TRUSTED_NUMBER,))
mute(store, WATCHED_GROUP, NOW + 600)
assert await _run(store, transport, trusting, "!odcisz", number=TRUSTED_NUMBER) == UNMUTED
assert not is_muted(store, WATCHED_GROUP, NOW)
async def test_a_stranger_asking_to_unmute_is_refused_in_silence(
store: Store, transport: FakeTransport, config: Config
) -> None:
"""No reply on purpose: the group is muted, and a refusal is a reply."""
trusting = replace(config, trusted_inviters=(TRUSTED_NUMBER,))
mute(store, WATCHED_GROUP, NOW + 600)
assert await _run(store, transport, trusting, "!odcisz", number=STRANGER_NUMBER) is None
assert is_muted(store, WATCHED_GROUP, NOW)
def test_the_real_aliases_register_without_a_collision() -> None:
"""The pair is registered from aliases.json, alongside nothing else."""
registry = CommandRegistry()
register_mute_commands(registry)
assert {c.name for c in registry.commands} == {MUTE_NAME, "unmute"}
assert registry.resolve("!cisza") is not None
assert registry.resolve("!odcisz") is not None