signal-bot/tests/test_manage_album.py
Krzysztof kuhy Rudnicki 0b539fc327 feat: the management backend, absorbing the tagger
A second process rather than more routes on the bot's own app. This one can
rewrite .env, and the bot's port already serves /_selftest, which posts for
real into a real group; keeping them apart means the surface that can
reconfigure the bot only listens while somebody has the page open.

The tagger's seven routes move here unchanged, out of tool/ and into the
coverage gate -- which is the reason an unconditional ImportError once sat
in tagger.py through two releases. The media route is still indexed by
integer, and that is not a style choice: a filename from a client is a path,
and the directory holds private photographs. FastAPI refuses a name with 422
before the handler runs, which the tests assert.

guard.py is the part that was not asked for. Binding to loopback stops the
network, not the browser already running on this machine, and two attacks
reach a loopback port from an ordinary web page: a cross-origin form, which
can only send urlencoded, multipart or plain text, so requiring
application/json refuses it with no token to keep; and DNS rebinding, whose
request still carries the attacker's hostname in Host. Forty lines instead
of authentication, which is the right trade for something started by hand
and stopped again a few minutes later.

assets.py branches on whether the page has been built, because StaticFiles
raises at construction on a missing directory -- mounting it unconditionally
would take the app down on a fresh clone, and in CI, where pytest never runs
pnpm build. Both branches are tested; "not built" answers with the command.

The container restart goes through signal_bot.external, wrapped in
asyncio.to_thread: it is synchronous with a 60s default, and blocking the
loop would freeze the page showing the restart.

100% branch coverage across all eleven modules, no omits, no pragmas.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01416TegJ3sa8xL5qGGgrygP
2026-08-31 14:38:24 +02:00

119 lines
4.6 KiB
Python

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The tagger's seven routes, absorbed and still behaving the same."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from signal_bot.pack import PEOPLE_FILE, TAGS_FILE
from tests.manage_fixture import FIXTURE_GROUP, bot_is_down, client_for, deps_at
if TYPE_CHECKING:
from pathlib import Path
def test_the_state_carries_the_album_and_the_groups_to_pick_from(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
body = client.get("/api/album/state").json()
assert [entry["name"] for entry in body["files"]] == ["one.png", "two.png"]
assert body["known_groups"] == [{"id": FIXTURE_GROUP, "name": "alfa"}]
def test_tagging_is_offline_work_and_survives_an_unreachable_container(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, bot_is_down)) as client:
assert client.get("/api/album/state").json()["known_groups"] == []
def test_tagging_one_file_records_who_is_in_it(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.post("/api/album/tag", json={"file": "two.png", "keys": ["p1"]})
assert answer.json() == {"status": "ok"}
tags = json.loads((deps.album.directory / TAGS_FILE).read_text(encoding="utf-8"))
assert tags["two.png"] == ["p1"]
def test_tagging_a_file_the_album_does_not_have_is_a_404(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/tag", json={"file": "nope.png", "keys": []})
assert answer.status_code == 404
def test_the_summary_says_what_each_command_would_answer_with(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
lines = client.get("/api/album/summary").json()["lines"]
assert lines[0].startswith("== 2 items")
def test_media_is_served_by_position_in_the_album(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.get("/api/album/media/0")
assert answer.status_code == 200
assert answer.content.startswith(b"\x89PNG")
def test_an_index_outside_the_album_is_a_404_at_either_end(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
assert client.get("/api/album/media/-1").status_code == 404
assert client.get("/api/album/media/2").status_code == 404
def test_a_filename_cannot_be_asked_for_at_all(tmp_path: Path) -> None:
# The whole defence: the route takes an int, so FastAPI refuses a name
# before the handler runs. This directory holds private photographs, and
# a filename from a client is a path.
with client_for(deps_at(tmp_path)) as client:
# 422 from FastAPI's own coercion, before any handler code runs.
assert client.get("/api/album/media/one.png").status_code == 422
assert client.get("/api/album/media/..%2F..%2Fetc%2Fpasswd").status_code != 200
def test_the_roster_can_be_written_from_the_page(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.post(
"/api/album/setup",
json={
"groups": [FIXTURE_GROUP],
"people": [{"key": "p1", "triggers": ["!alfa"]}],
"combos": [],
},
)
assert answer.json() == {"status": "ok"}
people = json.loads((deps.album.directory / PEOPLE_FILE).read_text(encoding="utf-8"))
assert people["groups"] == [FIXTURE_GROUP]
def test_a_roster_the_bot_could_not_load_is_refused_at_the_form(
tmp_path: Path,
) -> None:
# Not found at the next restart: a manifest the bot would refuse should
# be fixed here, while the person who wrote it is still looking at it.
with client_for(deps_at(tmp_path)) as client:
answer = client.post(
"/api/album/setup",
json={"groups": [], "people": [{"key": "p1", "triggers": ["!a"]}], "combos": []},
)
assert answer.status_code == 400
def test_a_person_can_be_added_mid_session(tmp_path: Path) -> None:
# The reason to add somebody is usually a photo of them already on screen.
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/person", json={"triggers": ["!delta"]})
assert answer.json()["status"] == "ok"
assert answer.json()["key"].startswith("p")
def test_adding_a_person_with_no_triggers_is_refused(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/person", json={"triggers": []})
assert answer.status_code == 400