signal-bot/tests/test_pack_pending.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The approval queue: in, listed, out -- and into the album only whole."""

from __future__ import annotations

import json
import logging
import stat
from typing import TYPE_CHECKING

import pytest

from signal_bot.groups import ConfigError
from signal_bot.pack import MEDIA_DIR, PEOPLE_FILE, TAGS_FILE
from signal_bot.pack_custom import CUSTOM_FILE
from signal_bot.pack_load import load
from signal_bot.pack_pending import (
    PENDING_DIR,
    Pending,
    approve,
    discard,
    listing,
    names,
    queue,
)
from signal_bot.pack_store import ADDS_FILE
from tests import pack_fixture
from tests.conftest import WATCHED_GROUP

if TYPE_CHECKING:
    from pathlib import Path

ONE = Pending("one.jpg", ("p1",), ("wakacje",), "uuid-alice", 100, WATCHED_GROUP)
TWO = Pending("two.jpg", (), ("gory", "morze"), "uuid-bob", 50, WATCHED_GROUP)
BODY = b"\xff\xd8\xff\xe0one"


def _album(tmp_path: Path) -> Path:
    root = pack_fixture.build(tmp_path / "pack")
    queue(root, BODY, ONE)
    queue(root, b"\xff\xd8\xff\xe0two", TWO)
    return root


def test_queued_entries_come_back_oldest_first_with_their_bytes(tmp_path: Path) -> None:
    root = _album(tmp_path)
    assert listing(root) == [TWO, ONE]
    assert names(root) == frozenset({"one.jpg", "two.jpg"})
    assert (root / PENDING_DIR / "one.jpg").read_bytes() == BODY
    for path in (root / PENDING_DIR).iterdir():
        assert stat.S_IMODE(path.stat().st_mode) == 0o600


def test_an_album_with_no_queue_yet_has_nothing_waiting(tmp_path: Path) -> None:
    root = pack_fixture.build(tmp_path / "pack")
    assert listing(root) == []
    assert names(root) == frozenset()


def test_a_broken_or_orphaned_sidecar_is_skipped_with_a_warning(
    tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """The page must still show the other photos."""
    root = _album(tmp_path)
    (root / PENDING_DIR / "broken.jpg.json").write_text("{not json", encoding="utf-8")
    (root / PENDING_DIR / "short.jpg.json").write_text('{"name": "short.jpg"}', encoding="utf-8")
    (root / PENDING_DIR / "two.jpg").unlink()
    # A sidecar whose name is not its own filename could point the page anywhere.
    escaped = json.dumps(
        {**json.loads((root / PENDING_DIR / "one.jpg.json").read_text()), "name": "../people.json"}
    )
    (root / PENDING_DIR / "escaped.jpg.json").write_text(escaped, encoding="utf-8")
    with caplog.at_level(logging.WARNING, logger="signal_bot.pack_pending"):
        assert listing(root) == [ONE]
    assert caplog.text.count("skipping unreadable or orphaned queue entry") == 4


def test_discard_removes_both_files_and_tolerates_absence(tmp_path: Path) -> None:
    root = _album(tmp_path)
    discard(root, "one.jpg")
    discard(root, "never-queued.jpg")
    assert listing(root) == [TWO]
    assert not (root / PENDING_DIR / "one.jpg.json").exists()


# --------------------------------------------------------------------------
# Approval
# --------------------------------------------------------------------------


def test_approving_moves_the_file_into_the_album_and_records_who_sent_it(
    tmp_path: Path,
) -> None:
    root = _album(tmp_path)
    assert approve(root, "one.jpg", ("p1",), (), now=200) == ("p1",)
    assert (root / MEDIA_DIR / "one.jpg").read_bytes() == BODY
    assert json.loads((root / TAGS_FILE).read_text(encoding="utf-8"))["one.jpg"] == ["p1"]
    adds = json.loads((root / ADDS_FILE).read_text(encoding="utf-8"))
    assert adds["one.jpg"] == {"by": "uuid-alice", "at": 100, "approved_at": 200}
    assert listing(root) == [TWO]
    assert "one.jpg" in load(root).matching(frozenset({"p1"}))


def test_a_proposed_label_is_minted_on_approval_and_reused_after(tmp_path: Path) -> None:
    root = _album(tmp_path)
    assert approve(root, "one.jpg", ("p1",), ("wakacje",), now=200) == ("p1", "t1")
    assert json.loads((root / CUSTOM_FILE).read_text(encoding="utf-8")) == {"t1": ["wakacje"]}

    queue(root, b"\xff\xd8\xff\xe0three", Pending("three.jpg", (), ("wakacje",), "u", 1, "g"))
    assert approve(root, "three.jpg", (), ("wakacje",), now=201) == ("t1",)
    assert load(root).matching(frozenset({"t1"})) == ("one.jpg", "three.jpg")


def test_the_approver_may_keep_fewer_labels_than_were_proposed(tmp_path: Path) -> None:
    root = _album(tmp_path)
    assert approve(root, "two.jpg", (), ("morze",), now=200) == ("t1",)
    assert json.loads((root / CUSTOM_FILE).read_text(encoding="utf-8")) == {"t1": ["morze"]}


def test_a_label_nobody_proposed_is_refused_before_anything_is_written(
    tmp_path: Path,
) -> None:
    root = _album(tmp_path)
    with pytest.raises(ValueError, match="not proposed"):
        approve(root, "one.jpg", ("p1",), ("wakacje", "wymyslona"), now=200)
    assert not (root / MEDIA_DIR / "one.jpg").exists()
    assert not (root / CUSTOM_FILE).exists()
    assert listing(root) == [TWO, ONE]


def test_a_name_that_is_not_waiting_is_a_key_error(tmp_path: Path) -> None:
    root = _album(tmp_path)
    with pytest.raises(KeyError, match="not waiting"):
        approve(root, "ghost.jpg", (), (), now=200)


def test_an_album_that_no_longer_loads_after_the_write_is_rolled_back(
    tmp_path: Path,
) -> None:
    """A key the album has no person for: written, unloadable, undone -- and still queued."""
    root = _album(tmp_path)
    tags_before = (root / TAGS_FILE).read_bytes()
    with pytest.raises(ConfigError):
        approve(root, "one.jpg", ("p99",), (), now=200)
    assert not (root / MEDIA_DIR / "one.jpg").exists()
    assert (root / TAGS_FILE).read_bytes() == tags_before
    assert not (root / CUSTOM_FILE).exists()
    assert not (root / ADDS_FILE).exists()
    assert listing(root) == [TWO, ONE]


def test_an_album_that_does_not_load_to_begin_with_writes_nothing(tmp_path: Path) -> None:
    root = _album(tmp_path)
    (root / PEOPLE_FILE).write_text("{broken", encoding="utf-8")
    with pytest.raises(ConfigError):
        approve(root, "one.jpg", ("p1",), (), now=200)
    assert not (root / MEDIA_DIR / "one.jpg").exists()
    assert listing(root) == [TWO, ONE]


def test_rows_the_loader_drops_survive_an_approval(tmp_path: Path) -> None:
    """tags.json is edited as a document: the tagger's "never opened" rows must stay."""
    root = _album(tmp_path)
    document = json.loads((root / TAGS_FILE).read_text(encoding="utf-8"))
    document["alfa.png"] = []
    (root / TAGS_FILE).write_text(json.dumps(document), encoding="utf-8")
    approve(root, "one.jpg", ("p1",), (), now=200)
    after = json.loads((root / TAGS_FILE).read_text(encoding="utf-8"))
    assert after["alfa.png"] == []
    assert after["one.jpg"] == ["p1"]


def test_an_album_the_tagger_has_not_reached_gets_its_first_tags_file(
    tmp_path: Path,
) -> None:
    root = pack_fixture.build(tmp_path / "pack", write_tags=False)
    queue(root, BODY, ONE)
    approve(root, "one.jpg", ("p1",), (), now=200)
    assert json.loads((root / TAGS_FILE).read_text(encoding="utf-8")) == {"one.jpg": ["p1"]}