mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-11 19:43:49 +01:00
Krzysztof kuhy Rudnicki
c17d50959b
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""What the dispatcher records when a gate refuses -- which is nothing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from typing import TYPE_CHECKING
|
|
|
|
from signal_bot.decorators import cooldown
|
|
from signal_bot.dispatcher import Dispatcher
|
|
from signal_bot.registry import CommandContext, CommandRegistry
|
|
from tests.conftest import WATCHED_GROUP, envelope
|
|
|
|
if TYPE_CHECKING:
|
|
from signal_bot.config import Config
|
|
from signal_bot.discovery import GroupDirectory
|
|
from signal_bot.store import Store
|
|
from signal_bot.transport_fake import FakeTransport
|
|
|
|
NOW = dt.datetime(2026, 8, 29, 12, 0, tzinfo=dt.UTC)
|
|
|
|
|
|
async def test_a_refused_command_records_nothing(
|
|
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
|
|
) -> None:
|
|
"""A gate's refusal must not slide the cooldown window forward.
|
|
|
|
This is what made "BEEP BOOP POCZEKAJ n SEKUND" a lie: the refusal itself
|
|
was recorded as a use, so waiting the stated n only bought another refusal.
|
|
"""
|
|
runs: list[str] = []
|
|
|
|
async def handler(ctx: CommandContext) -> None:
|
|
runs.append(ctx.command_name)
|
|
|
|
registry = CommandRegistry()
|
|
registry.register("cat", ["!cat"], cooldown(10, "wait {seconds}")(handler))
|
|
dispatcher = Dispatcher(
|
|
config=config,
|
|
directory=directory,
|
|
registry=registry,
|
|
store=store,
|
|
transport=transport,
|
|
clock=lambda: NOW,
|
|
)
|
|
earlier = NOW.timestamp() - 4.0
|
|
store.record_use(WATCHED_GROUP, "uuid-alice", "cat", earlier, NOW.date().isoformat())
|
|
|
|
await dispatcher.handle(envelope(text="!cat"))
|
|
|
|
assert runs == []
|
|
assert store.last_used(WATCHED_GROUP, "uuid-alice", "cat") == earlier
|
|
# The seeded run is the only one counted; a refusal must not eat an
|
|
# allowance either, which is the same bug in daily_limit.
|
|
assert store.uses_today(WATCHED_GROUP, "uuid-alice", "cat", NOW.date().isoformat()) == 1
|
|
|
|
|
|
async def test_waiting_the_stated_time_lets_the_command_run(
|
|
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
|
|
) -> None:
|
|
"""The user's own loop: obey the reply, retry, and it must work."""
|
|
runs: list[str] = []
|
|
clock = [NOW]
|
|
|
|
async def handler(ctx: CommandContext) -> None:
|
|
runs.append(ctx.command_name)
|
|
|
|
registry = CommandRegistry()
|
|
registry.register("cat", ["!cat"], cooldown(10, "{seconds}")(handler))
|
|
dispatcher = Dispatcher(
|
|
config=config,
|
|
directory=directory,
|
|
registry=registry,
|
|
store=store,
|
|
transport=transport,
|
|
clock=lambda: clock[0],
|
|
)
|
|
|
|
await dispatcher.handle(envelope(text="!cat"))
|
|
clock[0] = NOW + dt.timedelta(seconds=3)
|
|
await dispatcher.handle(envelope(text="!cat"))
|
|
assert runs == ["cat"]
|
|
|
|
waited = int(transport.texts[-1])
|
|
clock[0] += dt.timedelta(seconds=waited)
|
|
await dispatcher.handle(envelope(text="!cat"))
|
|
assert runs == ["cat", "cat"]
|