mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:07:19 +01:00
Krzysztof kuhy Rudnicki
da016e1b94
A photo sent to the chat no longer enters the album by itself. pack_add strips it, checks it against the album and the queue, and writes it to pending/ with a sidecar naming the people and labels the sender proposed; the reply is "czeka na zatwierdzenie". The management page lists the queue above the tagger, with the proposal pre-ticked, and Approve moves the file in, mints the labels, records approved_at and asks the bot to rebuild its table through /_reload -- the same publish path every other data change takes, so PackHolder.reload is gone. A reload the bot refuses is reported on the page, not rolled back: the photo is in the album either way. The personal-data gate now reads pending/*.json too, so a label that is still waiting for approval is forbidden in the tree already. The e2e run boots the management page beside the bot with the same environment, proves the queue is stripped and never reaches media/, approves it through the real routes, and has !losowe answer with the photo afterwards without a restart; exiftool is found through media_strip.locate() there rather than PATH, which lacks vendor_perl on Arch. !gadaj is dropped from !odcisz: it was within a typo of !dodaj, and the nearest-command suggestion must never be a coin toss. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
251 lines
10 KiB
Python
251 lines
10 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""!add_photo: what lands in the queue, and what is refused before it can."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from signal_bot.pack import MEDIA_DIR, TAGS_FILE
|
|
from signal_bot.pack_custom import CUSTOM_FILE
|
|
from signal_bot.pack_pending import PENDING_DIR, listing
|
|
from signal_bot.pack_replies import (
|
|
AMBIGUOUS,
|
|
DUPLICATE_ONLY,
|
|
NEED_LABEL,
|
|
NEED_PHOTO,
|
|
QUEUED_ONLY,
|
|
REJECTED,
|
|
TOO_MANY,
|
|
)
|
|
from signal_bot.pack_store import ADDS_FILE, content_name
|
|
from tests.conftest import SECOND_GROUP, WATCHED_GROUP
|
|
from tests.pack_add_fixture import (
|
|
JPEG_ONE,
|
|
JPEG_TWO,
|
|
NOT_MEDIA,
|
|
FakeStripper,
|
|
Wiring,
|
|
_run,
|
|
holder,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from signal_bot.config import Config
|
|
from signal_bot.pack_holder import PackHolder
|
|
from signal_bot.store import Store
|
|
from signal_bot.transport_fake import FakeTransport
|
|
|
|
__all__ = ["holder"]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# The happy paths: everything goes to the queue, nothing to the album
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_photo_tagged_with_a_person_waits_in_the_queue(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
before = len(holder.pack.matching(frozenset({"p1"})))
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1",))
|
|
(entry,) = listing(holder.pack.directory)
|
|
assert entry.keys == ("p1",)
|
|
assert entry.labels == ()
|
|
assert entry.by == "uuid-alice"
|
|
assert entry.at == 1756600000
|
|
assert entry.group == WATCHED_GROUP
|
|
assert (holder.pack.directory / PENDING_DIR / entry.name).read_bytes() == JPEG_ONE
|
|
assert transport.texts[-1] == "czeka na zatwierdzenie: 1"
|
|
assert len(holder.pack.matching(frozenset({"p1"}))) == before
|
|
|
|
|
|
async def test_an_unknown_label_is_proposed_not_minted(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
"""custom.json must not appear: the label is minted when the photo is approved."""
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj wakacje", attachments=("a1",))
|
|
(entry,) = listing(holder.pack.directory)
|
|
assert entry.keys == ()
|
|
assert entry.labels == ("wakacje",)
|
|
assert not (holder.pack.directory / CUSTOM_FILE).exists()
|
|
|
|
|
|
async def test_two_labels_propose_both(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa wakacje", attachments=("a1",))
|
|
(entry,) = listing(holder.pack.directory)
|
|
assert (entry.keys, entry.labels) == (("p1",), ("wakacje",))
|
|
|
|
|
|
async def test_a_whole_batch_is_queued(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
"""Picking three photos at once is the normal phone gesture."""
|
|
transport.payloads.update({"a1": JPEG_ONE, "a2": JPEG_TWO})
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1", "a2"))
|
|
assert len(listing(holder.pack.directory)) == 2
|
|
assert transport.texts[-1] == "czeka na zatwierdzenie: 2"
|
|
|
|
|
|
async def test_a_combination_trigger_proposes_everybody_in_it(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj oboje", attachments=("a1",))
|
|
(entry,) = listing(holder.pack.directory)
|
|
assert entry.keys == ("p1", "p2")
|
|
|
|
|
|
async def test_the_album_manifests_are_untouched_by_a_send(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
root = holder.pack.directory
|
|
tags_before = (root / TAGS_FILE).read_text(encoding="utf-8")
|
|
media_before = sorted(p.name for p in (root / MEDIA_DIR).iterdir())
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa wakacje", attachments=("a1",))
|
|
assert (root / TAGS_FILE).read_text(encoding="utf-8") == tags_before
|
|
assert sorted(p.name for p in (root / MEDIA_DIR).iterdir()) == media_before
|
|
assert not (root / ADDS_FILE).exists()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# The refusals, none of which may write anything
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_no_label_queues_nothing(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj", attachments=("a1",))
|
|
assert transport.texts[-1] == NEED_LABEL
|
|
assert listing(holder.pack.directory) == []
|
|
|
|
|
|
async def test_no_attachment_changes_nothing_on_disk(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
root = holder.pack.directory
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj piesek")
|
|
assert transport.texts[-1] == NEED_PHOTO
|
|
assert not (root / PENDING_DIR).exists()
|
|
assert not (root / CUSTOM_FILE).exists()
|
|
|
|
|
|
async def test_too_many_labels_are_refused(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
labels = " ".join(f"etykieta{n}" for n in range(7))
|
|
await _run(Wiring(holder, transport, store, config), f"!dodaj {labels}", attachments=("a1",))
|
|
assert transport.texts[-1] == TOO_MANY
|
|
|
|
|
|
async def test_an_either_or_combination_is_refused_rather_than_guessed(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
""" "!ktokolwiek" means two different answers; tagging cannot pick one."""
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj ktokolwiek", attachments=("a1",))
|
|
assert transport.texts[-1] == AMBIGUOUS
|
|
|
|
|
|
async def test_a_file_that_is_not_media_is_refused_on_its_bytes(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = NOT_MEDIA
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1",))
|
|
assert transport.texts[-1] == REJECTED
|
|
assert not (holder.pack.directory / PENDING_DIR).exists()
|
|
|
|
|
|
async def test_a_file_the_stripper_refuses_is_not_queued(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(
|
|
Wiring(holder, transport, store, config),
|
|
"!dodaj alfa",
|
|
stripper=FakeStripper(accepts=False),
|
|
attachments=("a1",),
|
|
)
|
|
assert transport.texts[-1] == REJECTED
|
|
assert listing(holder.pack.directory) == []
|
|
|
|
|
|
async def test_an_attachment_the_container_lost_is_refused(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("missing",))
|
|
assert transport.texts[-1] == REJECTED
|
|
|
|
|
|
async def test_it_does_not_exist_in_another_group(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(
|
|
Wiring(holder, transport, store, config),
|
|
"!dodaj alfa",
|
|
attachments=("a1",),
|
|
group=SECOND_GROUP,
|
|
)
|
|
assert transport.sent == []
|
|
assert transport.deleted == []
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Duplicates: of the album, and of the queue
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_photo_already_in_the_album_is_a_duplicate_even_when_untagged(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
"""Checked against media/ on disk, not the tag table the tagger has not reached."""
|
|
(holder.pack.directory / MEDIA_DIR / content_name(JPEG_ONE, ".jpg")).write_bytes(JPEG_ONE)
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1",))
|
|
assert transport.texts[-1] == DUPLICATE_ONLY
|
|
assert listing(holder.pack.directory) == []
|
|
|
|
|
|
async def test_a_photo_already_waiting_is_reported_as_waiting(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads["a1"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1",))
|
|
transport.payloads["a2"] = JPEG_ONE
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj beta", attachments=("a2",))
|
|
assert transport.texts[-1] == QUEUED_ONLY
|
|
(entry,) = listing(holder.pack.directory)
|
|
assert entry.keys == ("p1",)
|
|
|
|
|
|
async def test_a_mixed_batch_queues_the_new_ones_and_counts_the_rest(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
(holder.pack.directory / MEDIA_DIR / content_name(JPEG_ONE, ".jpg")).write_bytes(JPEG_ONE)
|
|
transport.payloads["a1"] = JPEG_TWO
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1",))
|
|
transport.payloads.update({"a2": JPEG_ONE, "a3": JPEG_TWO, "a4": b"\xff\xd8\xff\xe0three"})
|
|
await _run(
|
|
Wiring(holder, transport, store, config), "!dodaj beta", attachments=("a2", "a3", "a4")
|
|
)
|
|
assert transport.texts[-1] == "czeka na zatwierdzenie: 1 · 1 już było · 1 już czeka"
|
|
assert len(listing(holder.pack.directory)) == 2
|
|
|
|
|
|
async def test_the_same_photo_twice_in_one_message_counts_once(
|
|
holder: PackHolder, transport: FakeTransport, store: Store, config: Config
|
|
) -> None:
|
|
transport.payloads.update({"a1": JPEG_ONE, "a2": JPEG_ONE})
|
|
await _run(Wiring(holder, transport, store, config), "!dodaj alfa", attachments=("a1", "a2"))
|
|
assert len(listing(holder.pack.directory)) == 1
|
|
assert transport.texts[-1] == "czeka na zatwierdzenie: 1 · 1 już czeka"
|