signal-bot/tests/test_portraits.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The on-disk papal portrait cache."""

from __future__ import annotations

import base64
from pathlib import Path

import httpx
import pytest

from signal_bot.http_client import ImageFetcher, ImageFetchError
from signal_bot.portraits import PortraitCache, default_cache_dir
from signal_bot.versioning import POPES, Pope

IMAGE = b"portrait-bytes"
PIUS = POPES[7]


def _names(directory: Path) -> list[str]:
    """List a directory. A helper so the async tests touch no pathlib."""
    return sorted(entry.name for entry in directory.iterdir())


def _truncate(path: Path) -> None:
    """Leave behind what a crash between create and write would leave."""
    path.write_bytes(b"")


def _mock_fetcher(seen: list[str] | None = None) -> ImageFetcher:
    """Build a fetcher whose Commons endpoint is a mock returning IMAGE."""

    def handler(request: httpx.Request) -> httpx.Response:
        if seen is not None:
            seen.append(str(request.url))
        if "commons" in str(request.url):
            page = {"imageinfo": [{"thumburl": "https://cdn/portrait.jpg"}]}
            return httpx.Response(200, json={"query": {"pages": [page]}})
        return httpx.Response(200, content=IMAGE)

    return ImageFetcher(httpx.AsyncClient(transport=httpx.MockTransport(handler)))


def _cache(tmp_path: Path, seen: list[str] | None = None) -> PortraitCache:
    """Build a cache writing into tmp_path."""
    return PortraitCache(_mock_fetcher(seen), "https://commons.example/w/api.php", tmp_path)


async def test_a_miss_downloads_and_writes_the_file(tmp_path: Path) -> None:
    seen: list[str] = []
    cache = _cache(tmp_path, seen)

    assert await cache.get_bytes(PIUS) == IMAGE

    assert _names(tmp_path) == [f"{PIUS.ordinal:03d}.jpg"]
    assert any("titles=File" in url for url in seen)


async def test_a_miss_leaves_no_temporary_file_behind(tmp_path: Path) -> None:
    await _cache(tmp_path).get_bytes(PIUS)
    assert [name for name in _names(tmp_path) if name.endswith(".tmp")] == []


async def test_a_hit_makes_no_http_request(tmp_path: Path) -> None:
    seen: list[str] = []
    cache = _cache(tmp_path, seen)
    await cache.get_bytes(PIUS)
    seen.clear()

    assert await cache.get_bytes(PIUS) == IMAGE
    assert seen == []


async def test_an_empty_cached_file_is_refetched(tmp_path: Path) -> None:
    cache = _cache(tmp_path)
    # Serving this would post a broken attachment for as long as it lived.
    _truncate(cache.path_for(PIUS))

    assert await cache.get_bytes(PIUS) == IMAGE


async def test_an_unwritable_cache_still_serves_the_photo(tmp_path: Path) -> None:
    cache = PortraitCache(
        _mock_fetcher(),
        "https://commons.example/w/api.php",
        tmp_path / "nested" / "file.txt" / "deeper",
    )
    (tmp_path / "nested").mkdir()
    (tmp_path / "nested" / "file.txt").write_text("not a directory")

    assert await cache.get_bytes(PIUS) == IMAGE


async def test_get_base64_encodes_what_get_bytes_returns(tmp_path: Path) -> None:
    encoded = await _cache(tmp_path).get_base64(PIUS)
    assert base64.b64decode(encoded) == IMAGE


def test_the_filename_comes_from_the_ordinal_not_from_commons(tmp_path: Path) -> None:
    awkward = Pope(
        ordinal=9,
        name="Test",
        title="Test",
        reign="1-2",
        portrait="../../etc/passwd, and a space (v2).PNG",
        latin="Testus",
    )
    assert _cache(tmp_path).path_for(awkward) == tmp_path / "009.png"


def test_an_unknown_suffix_is_refused_rather_than_guessed(tmp_path: Path) -> None:
    weird = Pope(ordinal=9, name="T", title="T", reign="1-2", portrait="x.svg", latin="T")
    with pytest.raises(ImageFetchError, match="unusable suffix"):
        _cache(tmp_path).path_for(weird)


def test_every_committed_portrait_has_a_suffix_the_cache_accepts(tmp_path: Path) -> None:
    cache = _cache(tmp_path)
    for pope in POPES:
        assert cache.path_for(pope).parent == tmp_path


@pytest.fixture
def clean_env(monkeypatch: pytest.MonkeyPatch) -> pytest.MonkeyPatch:
    """A monkeypatch with every cache-directory variable removed."""
    for key in ("PORTRAIT_CACHE_DIR", "CACHE_DIRECTORY", "XDG_CACHE_HOME"):
        monkeypatch.delenv(key, raising=False)
    return monkeypatch


def test_an_explicit_directory_wins(clean_env: pytest.MonkeyPatch, tmp_path: Path) -> None:
    clean_env.setenv("PORTRAIT_CACHE_DIR", str(tmp_path / "explicit"))
    clean_env.setenv("CACHE_DIRECTORY", str(tmp_path / "systemd"))
    assert default_cache_dir() == tmp_path / "explicit"


def test_systemd_cache_directory_is_used_when_set(
    clean_env: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Under the unit this is the only writable path: ProtectHome=read-only."""
    clean_env.setenv("CACHE_DIRECTORY", str(tmp_path / "systemd"))
    assert default_cache_dir() == tmp_path / "systemd" / "portraits"


def test_xdg_is_the_fallback_outside_systemd(clean_env: pytest.MonkeyPatch, tmp_path: Path) -> None:
    clean_env.setenv("XDG_CACHE_HOME", str(tmp_path / "xdg"))
    assert default_cache_dir() == tmp_path / "xdg" / "signal-bot" / "portraits"


def test_home_is_the_last_resort(clean_env: pytest.MonkeyPatch, tmp_path: Path) -> None:
    clean_env.setattr(Path, "home", classmethod(lambda _: tmp_path / "home"))
    assert default_cache_dir() == tmp_path / "home" / ".cache" / "signal-bot" / "portraits"