mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:07:19 +01:00
The registry used to be built once, inline in build_bot, so every change to the data behind it -- a person tagged in the web UI, an alias, soon a command switched off from the management page -- meant restarting the unit. That is cheap and lossless, but it is a deploy step somebody has to remember, and a 21:37 report can land in the middle of it. Now the registration sequence lives in wiring.build_registry, and RegistryHolder runs it again on POST /_reload: the new table is built whole and swapped in with one assignment, never patched in place, so a duplicate trigger is found before anything the dispatcher reads has changed. Bad data (ConfigError, DuplicateTriggerError) keeps the previous table and answers 409 with the reason; a programming error still surfaces as a 500. scripts/reload.sh wraps it, with --code for the restart a Python change needs, and an optional command to type in testbot so "reloaded" means "answered". No Python hot-reload on purpose. The loader change is what makes deploying this safe: two rows in the live tags.json name files that are no longer in media/, and load_pack refused them, so the next restart would have looped on ConfigError. A stale tag is drift in a file a page edits mid-session, not a broken album -- it is now skipped with a warning naming it. A tag that reaches outside the album is still refused. pack_load.py split at the 250-line cap; pack_tags.py is the half whose checks are not all fatal. Proven offline in scripts/e2e.sh: !delta is unknown, a fifth person is tagged into the synthetic pack, scripts/reload.sh runs, and the same process posts delta's photo. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
204 lines
8.0 KiB
Python
204 lines
8.0 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
||
"""Everything a photo pack can get wrong, and the sentence it gets back.
|
||
|
||
Each of these fails startup rather than a command. A manifest that cannot be
|
||
honoured is an answer, not an outage: the alternative is a bot that comes up
|
||
healthy and answers "no such photo" forever because a file was renamed.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import TYPE_CHECKING
|
||
|
||
import pytest
|
||
|
||
from signal_bot.groups import ConfigError
|
||
from signal_bot.pack_load import load
|
||
from tests import pack_fixture
|
||
|
||
if TYPE_CHECKING:
|
||
from pathlib import Path
|
||
|
||
from signal_bot.pack import Pack
|
||
|
||
|
||
def _load(root: Path, **kwargs: object) -> Pack:
|
||
"""Build a pack of the requested shape and load it."""
|
||
return load(pack_fixture.build(root / "pack", **kwargs))
|
||
|
||
|
||
def test_a_pack_loads_its_people_its_combos_and_its_tags(tmp_path: Path) -> None:
|
||
pack = _load(tmp_path)
|
||
assert [entry.key for entry in pack.entries] == ["p1", "p2", "p3", "c1", "c2"]
|
||
assert pack.tags["both.png"] == frozenset({"p1", "p2"})
|
||
assert pack.groups
|
||
|
||
|
||
def test_a_missing_manifest_says_which_file(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="cannot be read"):
|
||
load(tmp_path / "nowhere")
|
||
|
||
|
||
def test_a_manifest_that_is_not_json_says_so(tmp_path: Path) -> None:
|
||
root = pack_fixture.build(tmp_path / "pack")
|
||
(root / "people.json").write_text("{oops", encoding="utf-8")
|
||
with pytest.raises(ConfigError, match="not valid JSON"):
|
||
load(root)
|
||
|
||
|
||
def test_a_manifest_that_is_not_an_object_is_refused(tmp_path: Path) -> None:
|
||
root = pack_fixture.build(tmp_path / "pack")
|
||
(root / "people.json").write_text("[]", encoding="utf-8")
|
||
with pytest.raises(ConfigError, match="must be an object"):
|
||
load(root)
|
||
|
||
|
||
def test_a_pack_with_no_groups_is_refused(tmp_path: Path) -> None:
|
||
"""Defaulting to "every group" would put private photos in a stranger's chat."""
|
||
with pytest.raises(ConfigError, match="non-empty 'groups'"):
|
||
_load(tmp_path, groups=[])
|
||
|
||
|
||
def test_a_blank_group_id_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="blank group id"):
|
||
_load(tmp_path, groups=[" "])
|
||
|
||
|
||
def test_people_have_to_be_objects(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="must be a list of objects"):
|
||
_load(tmp_path, people=["alfa"])
|
||
|
||
|
||
def test_a_pack_with_no_entries_at_all_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="defines no commands"):
|
||
_load(tmp_path, people=[], combos=[], tags={})
|
||
|
||
|
||
def test_a_key_that_reads_like_a_name_is_refused(tmp_path: Path) -> None:
|
||
"""The key is what gets logged; it exists precisely so a name is not."""
|
||
with pytest.raises(ConfigError, match="alphanumeric 'key'"):
|
||
_load(tmp_path, people=[{"key": "", "triggers": ["!alfa"]}], combos=[], tags={})
|
||
|
||
|
||
def test_a_non_ascii_key_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="alphanumeric 'key'"):
|
||
_load(tmp_path, people=[{"key": "gamma\u0105", "triggers": ["!alfa"]}], combos=[], tags={})
|
||
|
||
|
||
def test_an_entry_needs_at_least_one_trigger(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="non-empty 'triggers'"):
|
||
_load(tmp_path, people=[{"key": "p1", "triggers": []}], combos=[], tags={})
|
||
|
||
|
||
def test_a_trigger_without_a_bang_could_never_fire(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match='must start with "!"'):
|
||
_load(tmp_path, people=[{"key": "p1", "triggers": ["alfa"]}], combos=[], tags={})
|
||
|
||
|
||
def test_a_trigger_with_a_digit_is_refused(tmp_path: Path) -> None:
|
||
"""folding maps 0134578 onto letters, so "!alfa7" is not what you typed."""
|
||
with pytest.raises(ConfigError, match="must not contain a digit"):
|
||
_load(tmp_path, people=[{"key": "p1", "triggers": ["!alfa7"]}], combos=[], tags={})
|
||
|
||
|
||
def test_a_trigger_that_folds_away_to_nothing_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="folds away to nothing"):
|
||
_load(tmp_path, people=[{"key": "p1", "triggers": ["!"]}], combos=[], tags={})
|
||
|
||
|
||
def test_a_combo_needs_exactly_one_of_all_of_and_any_of(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="exactly one of"):
|
||
_load(
|
||
tmp_path,
|
||
combos=[{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1"], "any_of": ["p2"]}],
|
||
)
|
||
|
||
|
||
def test_a_combo_with_neither_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="exactly one of"):
|
||
_load(tmp_path, combos=[{"key": "c1", "triggers": ["!oboje"]}])
|
||
|
||
|
||
def test_a_combo_naming_somebody_who_does_not_exist_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="unknown people"):
|
||
_load(tmp_path, combos=[{"key": "c1", "triggers": ["!oboje"], "all_of": ["p9"]}])
|
||
|
||
|
||
def test_a_key_used_twice_is_refused(tmp_path: Path) -> None:
|
||
"""Two entries on one key would share a cooldown row and a log line."""
|
||
people = [{"key": "p1", "triggers": ["!alfa"]}, {"key": "p1", "triggers": ["!beta"]}]
|
||
with pytest.raises(ConfigError, match="twice"):
|
||
_load(tmp_path, people=people, combos=[], tags={})
|
||
|
||
|
||
def test_two_entries_folding_onto_one_trigger_are_refused(tmp_path: Path) -> None:
|
||
"""The registry would catch it too, but could not say which entries."""
|
||
people = [{"key": "p1", "triggers": ["!gamma"]}, {"key": "p2", "triggers": ["!gammą"]}]
|
||
with pytest.raises(ConfigError, match="claimed by"):
|
||
_load(tmp_path, people=people, combos=[], tags={})
|
||
|
||
|
||
def test_one_entry_may_spell_itself_two_ways_that_fold_the_same(tmp_path: Path) -> None:
|
||
"""Which is the point: "!gamma" and "!gammą" are one person, written twice."""
|
||
pack = _load(
|
||
tmp_path,
|
||
people=[{"key": "p1", "triggers": ["!gamma", "!gammą"]}],
|
||
combos=[],
|
||
tags={},
|
||
)
|
||
assert len(pack.entries) == 1
|
||
|
||
|
||
def test_an_untagged_pack_loads_and_warns(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||
"""Not an error: that is an album waiting to be tagged, not a broken one."""
|
||
pack = _load(tmp_path, write_tags=False)
|
||
assert pack.tags == {}
|
||
# The warning names the command to run, so it has to move with it.
|
||
assert "./web.sh" in caplog.text
|
||
|
||
|
||
def test_a_tag_naming_a_file_that_is_gone_is_skipped_out_loud(
|
||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||
) -> None:
|
||
"""Drift in one row must not take the album down; it is named, then ignored.
|
||
|
||
Fatal until 2026-09-04, when two such rows in the live pack meant the
|
||
next restart would have looped on ConfigError.
|
||
"""
|
||
pack = _load(
|
||
tmp_path,
|
||
tags={"alfa.png": ["p1"], "gone.jpg": ["p2"]},
|
||
media={"alfa.png": pack_fixture.PIXEL},
|
||
)
|
||
assert set(pack.tags) == {"alfa.png"}
|
||
assert "'gone.jpg', which is not a file in media/ -- skipped" in caplog.text
|
||
|
||
|
||
def test_a_tag_may_not_reach_outside_the_album(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="not a file"):
|
||
_load(tmp_path, tags={"../people.json": ["p1"]}, media={"alfa.png": pack_fixture.PIXEL})
|
||
|
||
|
||
def test_a_tag_on_a_type_signal_is_not_told_about_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match="not one Signal is told about"):
|
||
_load(tmp_path, tags={"notes.txt": ["p1"]}, media={"notes.txt": b"hello"})
|
||
|
||
|
||
def test_a_tag_naming_somebody_outside_the_manifest_is_refused(tmp_path: Path) -> None:
|
||
with pytest.raises(ConfigError, match=r"names somebody who is not in"):
|
||
_load(tmp_path, tags={"alfa.png": ["p9"]})
|
||
|
||
|
||
def test_a_tag_that_is_not_a_list_is_refused(tmp_path: Path) -> None:
|
||
root = pack_fixture.build(tmp_path / "pack")
|
||
(root / "tags.json").write_text(json.dumps({"alfa.png": "p1"}), encoding="utf-8")
|
||
with pytest.raises(ConfigError, match="names somebody who is not in"):
|
||
load(root)
|
||
|
||
|
||
def test_an_untagged_file_is_simply_not_in_the_album(tmp_path: Path) -> None:
|
||
"""The tagger writes an empty list for "seen, nobody in it"."""
|
||
pack = _load(tmp_path, tags={"alfa.png": [], "both.png": ["p1", "p2"]})
|
||
assert set(pack.tags) == {"both.png"}
|