signal-bot/tool/e2e_pack.py
Krzysztof kuhy Rudnicki 0717df646a
Some checks are pending
CI / check (push) Waiting to run
CI / web (push) Waiting to run
Rebuild the command table on demand, and survive a tag whose file is gone
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>
2026-09-04 10:14:39 +02:00

80 lines
3.0 KiB
Python
Executable File

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Write a synthetic photo pack for the offline end-to-end run.
The real album is private and lives outside this repository, so the e2e
script cannot use it -- and must not need to. This builds a pack of three
invented people, one "both of them" combination and one video, which is
enough to exercise every path the real one uses.
python tool/e2e_pack.py --dir /tmp/pack --group 'group.abc='
"""
from __future__ import annotations
import argparse
import base64
import json
import sys
from pathlib import Path
from typing import Final
# A 1x1 PNG, inline so that no binary file enters the repository.
PIXEL: Final = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)
# Not playable, and it does not have to be: nothing decodes it, the point is
# that it leaves as video/mp4 rather than as a photo.
CLIP: Final = b"\x00\x00\x00\x18ftypmp42fake"
PEOPLE: Final = [
{"key": "p1", "triggers": ["!alfa"]},
{"key": "p2", "triggers": ["!beta"]},
{"key": "p3", "triggers": ["!gamma"]},
{"key": "p4", "triggers": ["!wideo"]},
]
COMBOS: Final = [{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1", "p2"]}]
TAGS: Final = {
"alfa.png": ["p1"],
"both.png": ["p1", "p2"],
"gamma.png": ["p3"],
"clip.mp4": ["p4"],
}
def write(directory: Path, group: str, extra: str | None = None) -> None:
"""Write the manifest, the tags and the files themselves.
``extra`` adds one more person, ``!<extra>`` with one photo, on top of
the fixed four -- run a second time over the same directory, that is
what "somebody tagged a new person while the bot was up" looks like,
and what POST /_reload has to pick up without a restart.
"""
people, tags = list(PEOPLE), dict(TAGS)
if extra is not None:
people.append({"key": "p5", "triggers": [f"!{extra}"]})
tags[f"{extra}.png"] = ["p5"]
album = directory / "media"
album.mkdir(parents=True, exist_ok=True)
for name in tags:
(album / name).write_bytes(CLIP if name.endswith(".mp4") else PIXEL)
manifest = {"groups": [group], "people": people, "combos": COMBOS}
(directory / "people.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
(directory / "tags.json").write_text(json.dumps(tags, indent=2), encoding="utf-8")
def main() -> int:
"""Build the pack asked for on the command line."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--dir", required=True, type=Path, help="where to write the pack")
parser.add_argument("--group", required=True, help="the group id the pack belongs to")
parser.add_argument("--extra", help="one more person to add, by trigger word")
arguments = parser.parse_args()
write(arguments.dir, arguments.group, arguments.extra)
sys.stdout.write(f"== wrote a synthetic pack to {arguments.dir}\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())