# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """The !pułapka command: a photograph of the device, with nothing attached.""" from __future__ import annotations import base64 import json from typing import TYPE_CHECKING import httpx import pytest from signal_bot.context import CommandContext from signal_bot.decorators import DISPLAY_MARGIN_SECONDS, cooldown from signal_bot.http_client import ImageFetcher from signal_bot.models import parse_envelope from signal_bot.registry import CommandRefusedError, CommandRegistry from signal_bot.traps_command import ( COMMAND_NAME, TRAPS_FILE, load_traps, random_trap, register_trap_command, ) from tests.conftest import WATCHED_GROUP, envelope if TYPE_CHECKING: from pathlib import Path from signal_bot.config import Config from signal_bot.store import Store from signal_bot.transport_fake import FakeTransport IMAGE = b"trap-bytes" API = "https://commons.example/w/api.php" ALIASES = {COMMAND_NAME: ["!pułapka", "!pulapka", "!trap"]} POOL = (111419226, 124481655, 151918804) def _fetcher(seen: list[httpx.URL] | None = None) -> ImageFetcher: def handler(request: httpx.Request) -> httpx.Response: if "commons" in str(request.url): if seen is not None: seen.append(request.url) page = { "title": "File:Molchreuse.JPG", "imageinfo": [ { "thumburl": "https://cdn/trap.jpg", "descriptionurl": "https://commons.wikimedia.org/wiki/File:Molchreuse.JPG", "extmetadata": { "ImageDescription": {"value": "Eine Molchreuse"}, }, } ], } return httpx.Response(200, json={"query": {"pages": [page]}}) return httpx.Response(200, content=IMAGE) return ImageFetcher(httpx.AsyncClient(transport=httpx.MockTransport(handler))) def _ctx(transport: FakeTransport, store: Store, config: Config, text: str) -> CommandContext: message = parse_envelope(envelope(text=text)) assert message is not None return CommandContext( message=message, group_id=WATCHED_GROUP, transport=transport, store=store, config=config, command_name=COMMAND_NAME, now=100.0, day="2026-08-31", ) @pytest.mark.parametrize("trigger", ["!pułapka", "!pulapka", "!trap"]) def test_every_alias_resolves(trigger: str) -> None: registry = CommandRegistry() register_trap_command(registry, _fetcher(), API, aliases=ALIASES, traps=POOL) command = registry.resolve(trigger) assert command is not None assert command.name == COMMAND_NAME def test_the_real_alias_file_carries_the_command() -> None: registry = CommandRegistry() register_trap_command(registry, _fetcher(), API, traps=POOL) # The Polish spelling is the primary one !pomoc prints; the English # !trap is the word that prompted the command in the first place. assert registry.resolve("!pułapka") is not None assert registry.resolve("!trap") is not None async def test_it_posts_an_image_captioned_with_what_the_thing_is( transport: FakeTransport, store: Store, config: Config ) -> None: registry = CommandRegistry() register_trap_command(registry, _fetcher(), API, aliases=ALIASES, traps=POOL) command = registry.resolve("!trap") assert command is not None await command.handler(_ctx(transport, store, config, "!trap")) sent = transport.sent[0] assert sent.recipient == WATCHED_GROUP assert sent.image_base64 is not None assert base64.b64decode(sent.image_base64) == IMAGE assert len(transport.sent) == 1 # Commons is full of files whose subject is unguessable, so the caption # says what it is and links the page that says more. Tags come out. assert sent.caption == ( "Eine Molchreuse\nhttps://commons.wikimedia.org/wiki/File:Molchreuse.JPG" ) async def test_it_names_the_file_by_page_id_and_never_by_title( transport: FakeTransport, store: Store, config: Config ) -> None: # A title would be free text carrying people's names into the repository; # see load_traps. This is the assertion that keeps the query on page ids. seen: list[httpx.URL] = [] registry = CommandRegistry() register_trap_command(registry, _fetcher(seen), API, aliases=ALIASES, traps=POOL) command = registry.resolve("!trap") assert command is not None await command.handler(_ctx(transport, store, config, "!trap")) assert len(seen) == 1 assert seen[0].params["pageids"] in {str(page) for page in POOL} assert "titles" not in seen[0].params async def test_the_cooldown_decorator_applies( transport: FakeTransport, store: Store, config: Config ) -> None: registry = CommandRegistry() register_trap_command( registry, _fetcher(), API, decorate=cooldown(10, "WAIT {seconds}"), aliases=ALIASES, traps=POOL, ) store.record_use(WATCHED_GROUP, "uuid-alice", COMMAND_NAME, 100.0, "2026-08-31") command = registry.resolve("!trap") assert command is not None with pytest.raises(CommandRefusedError): await command.handler(_ctx(transport, store, config, "!trap")) assert transport.texts == [f"WAIT {10 + DISPLAY_MARGIN_SECONDS}"] def test_the_draw_actually_varies() -> None: """0.0.3 posted one identical gecko forever; that is what this catches.""" drawn = {random_trap(POOL) for _ in range(20)} assert len(drawn) > 1 def test_the_shipped_pool_is_page_ids_and_nothing_else() -> None: """The personal-data invariant, as a unit test. A filename here would carry photographers' and sitters' names into a public repository, and one of them colliding with a label somebody minted from the chat wedges check_personal_data.py permanently. Integers cannot. """ raw = json.loads(TRAPS_FILE.read_text(encoding="utf-8"))["traps"] assert raw assert all(isinstance(page, int) for page in raw) assert len(set(raw)) == len(raw) assert raw == sorted(raw) def test_the_shipped_pool_is_large_enough_to_be_worth_drawing_from() -> None: # kuhy asked for a set "as big as possible", and a regeneration that # silently truncated would still pass every other test here. assert len(load_traps()) >= 500 @pytest.mark.parametrize("traps", [[], ["111419226.jpg"], [None]]) def test_a_pool_that_is_not_page_ids_is_refused(tmp_path: Path, traps: list[object]) -> None: path = tmp_path / "traps.json" path.write_text(json.dumps({"traps": traps}), encoding="utf-8") with pytest.raises(ValueError, match="page ids"): load_traps(path)