# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """The cat and dog commands, and their data-driven aliases.""" from __future__ import annotations import json from typing import TYPE_CHECKING import httpx import pytest from signal_bot.animals import ( ALIAS_FILE, INATURALIST_PAGE_LIMIT, load_aliases, paged_url, register_animal_commands, ) 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 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"png-bytes" def _fetcher(seen: list[str] | None = None) -> ImageFetcher: """Build a fetcher over the three fixture APIs, optionally recording URLs.""" def handler(request: httpx.Request) -> httpx.Response: url = str(request.url) if seen is not None: seen.append(url) if "thecatapi" in url: return httpx.Response(200, json=[{"url": "https://cdn/cat.png"}]) if "dog.ceo" in url: return httpx.Response(200, json={"message": "https://cdn/dog.png"}) if "inaturalist" in url: photo = {"url": "https://cdn/photos/1/square.jpeg"} return httpx.Response(200, json={"results": [{"photos": [photo]}]}) return httpx.Response(200, content=IMAGE) return ImageFetcher(httpx.AsyncClient(transport=httpx.MockTransport(handler))) def _ctx( transport: FakeTransport, store: Store, config: Config, name: str, 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=name, now=100.0, day="2026-08-29", ) def test_alias_file_ships_with_the_package() -> None: assert ALIAS_FILE.is_file() aliases = load_aliases() assert "!cat" in aliases["cat"] assert "!pies" in aliases["dog"] assert "!gekon" in aliases["gecko"] def test_comment_keys_are_skipped(tmp_path: Path) -> None: path = tmp_path / "a.json" path.write_text(json.dumps({"_comment": "x", "cat": ["!cat"]}), encoding="utf-8") assert load_aliases(path) == {"cat": ["!cat"]} def test_styled_unicode_aliases_resolve() -> None: registry = CommandRegistry() register_animal_commands(registry, _fetcher()) assert registry.resolve("!\U0001d5f8\U0001d5fc\U0001d601") is not None assert registry.resolve("!miau").name == "cat" assert registry.resolve("!szczek").name == "dog" assert registry.resolve("!gekon").name == "gecko" @pytest.mark.parametrize( ("trigger", "name"), [("!cat", "cat"), ("!dog", "dog"), ("!gekon", "gecko")] ) async def test_commands_post_an_image( transport: FakeTransport, store: Store, config: Config, trigger: str, name: str ) -> None: registry = CommandRegistry() register_animal_commands(registry, _fetcher()) command = registry.resolve(trigger) assert command is not None await command.handler(_ctx(transport, store, config, name, trigger)) assert transport.sent[0].image_base64 is not None assert transport.sent[0].recipient == WATCHED_GROUP async def test_decorator_is_applied_when_given( transport: FakeTransport, store: Store, config: Config ) -> None: registry = CommandRegistry() register_animal_commands(registry, _fetcher(), decorate=cooldown(10, "wait {seconds}")) store.record_use(WATCHED_GROUP, "uuid-alice", "cat", 100.0, "2026-08-29") command = registry.resolve("!cat") assert command is not None with pytest.raises(CommandRefusedError): await command.handler(_ctx(transport, store, config, "cat", "!cat")) assert transport.texts == [f"wait {10 + DISPLAY_MARGIN_SECONDS}"] def test_custom_alias_table_is_honoured() -> None: registry = CommandRegistry() table = {"cat": ["!c"], "dog": ["!d"], "gecko": ["!g"]} register_animal_commands(registry, _fetcher(), aliases=table) assert registry.resolve("!c").name == "cat" assert registry.resolve("!cat") is None def test_paged_url_leaves_a_source_without_a_page_limit_alone() -> None: assert paged_url("https://api/x?a=1", None) == "https://api/x?a=1" @pytest.mark.parametrize( ("api_url", "separator"), [("https://api/x?a=1", "&"), ("https://api/x", "?")] ) def test_paged_url_appends_a_page_with_the_right_separator(api_url: str, separator: str) -> None: url = paged_url(api_url, 10) prefix = f"{api_url}{separator}page=" assert url.startswith(prefix) assert 1 <= int(url.removeprefix(prefix)) <= 10 def test_paged_url_actually_varies() -> None: """The regression: iNaturalist accepts order_by=random and ignores it. 0.0.3 shipped with that as the only source of variety and posted one identical gecko every time. Twenty draws over a 10 000-page range collide with probability far below any plausible flake budget. """ drawn = {paged_url("https://api/x?a=1", INATURALIST_PAGE_LIMIT) for _ in range(20)} assert len(drawn) > 1 async def test_gecko_asks_for_the_full_size_photo( transport: FakeTransport, store: Store, config: Config ) -> None: """iNaturalist names a 75px thumbnail; the group should get the real photo.""" seen: list[str] = [] registry = CommandRegistry() register_animal_commands(registry, _fetcher(seen)) command = registry.resolve("!gecko") assert command is not None await command.handler(_ctx(transport, store, config, "gecko", "!gecko")) assert "https://cdn/photos/1/large.jpeg" in seen assert "https://cdn/photos/1/square.jpeg" not in seen assert "page=" in seen[0] async def test_cat_is_asked_for_without_a_page( transport: FakeTransport, store: Store, config: Config ) -> None: """Only the sources that need paging get one; thecatapi is random already.""" seen: list[str] = [] registry = CommandRegistry() register_animal_commands(registry, _fetcher(seen)) command = registry.resolve("!cat") assert command is not None await command.handler(_ctx(transport, store, config, "cat", "!cat")) assert "page=" not in seen[0]