signal-bot/tests/test_decorators.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The reusable gates, tested against a harmless command."""

from __future__ import annotations

from dataclasses import replace
from typing import TYPE_CHECKING

import pytest

from signal_bot.decorators import cooldown, daily_limit, vote_gated
from signal_bot.models import parse_envelope
from signal_bot.registry import CommandContext, Handler
from tests.conftest import SECOND_GROUP, 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

DAY = "2026-08-29"


def _context(
    store: Store,
    transport: FakeTransport,
    config: Config,
    *,
    uuid: str = "u1",
    now: float = 100.0,
    group: str = WATCHED_GROUP,
) -> CommandContext:
    message = parse_envelope(envelope(text="!ping", uuid=uuid, group_id=group))
    assert message is not None
    return CommandContext(
        message=message,
        group_id=group,
        transport=transport,
        store=store,
        config=config,
        command_name="ping",
        now=now,
        day=DAY,
    )


@pytest.fixture
def runs() -> list[str]:
    return []


@pytest.fixture
def handler(runs: list[str]) -> Handler:
    async def _handler(ctx: CommandContext) -> None:
        runs.append(ctx.message.source_uuid)

    return _handler


async def test_cooldown_allows_first_use(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    await cooldown(10, "wait {seconds}")(handler)(_context(store, transport, config))
    assert runs == ["u1"]


async def test_cooldown_blocks_and_reports_remaining(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    ctx = _context(store, transport, config, now=104.0)
    await cooldown(10, "wait {seconds}")(handler)(ctx)
    assert runs == []
    assert transport.texts == ["wait 7"]


async def test_cooldown_expires(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    await cooldown(10, "wait {seconds}")(handler)(_context(store, transport, config, now=111.0))
    assert runs == ["u1"]


async def test_cooldown_is_per_user(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    """One person's cat picture used to silence everybody."""
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    await cooldown(10, "wait {seconds}")(handler)(
        _context(store, transport, config, uuid="u2", now=101.0)
    )
    assert runs == ["u2"]


async def test_daily_limit_allows_under_the_cap(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    await daily_limit(1, "no more")(handler)(_context(store, transport, config))
    assert runs == ["u1"]


async def test_daily_limit_blocks_at_the_cap(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    await daily_limit(1, "no more, used {used}/{max_uses}")(handler)(
        _context(store, transport, config)
    )
    assert runs == []
    assert transport.texts == ["no more, used 1/1"]


async def test_vote_gate_waits_for_quorum(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    gated = vote_gated(2, 900, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1"))
    assert runs == []
    assert transport.texts == ["1/2"]


async def test_vote_gate_fires_on_quorum(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    gated = vote_gated(2, 900, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1"))
    await gated(_context(store, transport, config, uuid="u2"))
    assert runs == ["u2"]


async def test_repeat_vote_does_not_reach_quorum(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    gated = vote_gated(2, 900, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1"))
    await gated(_context(store, transport, config, uuid="u1", now=101.0))
    assert runs == []


async def test_votes_expire_outside_the_window(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    gated = vote_gated(2, 60, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1", now=100.0))
    await gated(_context(store, transport, config, uuid="u2", now=1000.0))
    assert runs == []


async def test_vote_tally_resets_after_firing(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    gated = vote_gated(2, 900, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1"))
    await gated(_context(store, transport, config, uuid="u2"))
    assert store.count_votes(WATCHED_GROUP, "ping", 0.0) == 0


async def test_gates_stack(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    stacked = cooldown(10, "wait {seconds}")(daily_limit(2, "no more")(handler))
    await stacked(_context(store, transport, config))
    assert runs == ["u1"]
    assert replace(config, cooldown_seconds=10).cooldown_seconds == 10


async def test_cooldown_is_per_group(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    """A test-group command must not throttle the production group."""
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    await cooldown(10, "wait {seconds}")(handler)(
        _context(store, transport, config, now=101.0, group=SECOND_GROUP)
    )
    assert runs == ["u1"]
    assert transport.texts == []


async def test_daily_limit_is_per_group(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    store.record_use(WATCHED_GROUP, "u1", "ping", 100.0, DAY)
    await daily_limit(1, "no more")(handler)(_context(store, transport, config, group=SECOND_GROUP))
    assert runs == ["u1"]


async def test_votes_do_not_carry_between_groups(
    store: Store, transport: FakeTransport, config: Config, handler: Handler, runs: list[str]
) -> None:
    """Quorum in one group must not fire the command in another."""
    gated = vote_gated(2, 900, "{votes}/{required}")(handler)
    await gated(_context(store, transport, config, uuid="u1"))
    await gated(_context(store, transport, config, uuid="u2", group=SECOND_GROUP))
    assert runs == []
    assert transport.texts == ["1/2", "1/2"]