signal-bot/tests/test_mute_command.py

# 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