mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 18:07:29 +01:00
Krzysztof kuhy Rudnicki
72e90acb47
A custom alias is data the command table is built from, so it cannot be a live switch like the ones already on the tab. Instead the management page writes aliases_custom.json in the bot's state directory and then asks the bot to reload, in one step: whenever the answer is anything but `reloaded` -- a word another command has, a name no command has, a bot still connecting -- the file is restored byte for byte and the bot's own reason is shown next to the field. What is on disk is therefore always what the bot serves, and a typo cannot sit in the file poisoning every later reload. The registry appends the operator's words to the shipped list at registration, which keeps `!pomoc` on the shipped spelling and routes the new words through the same clash check as aliases.json. An alias for a name no command has fails the whole build (ConfigError -> `kept`), because a silently ignored entry is how a file drifts. The tagger treats custom aliases as reserved words too, so a person cannot be given one. registry.py crossed the 250-line cap and now re-exports the record and the error types from command.py; the bot's state directory lookup, previously duplicated in pack.py and the suggestions export, is state_dir.py. Proven end to end: !e2ekot is unknown, the file is written and reload.sh runs, !e2ekot posts a cat, a file naming a command that does not exist is refused with 409 and the previous table still answers. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""The alias editor: written to one file, made live by the bot, or undone."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import TYPE_CHECKING
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from signal_bot.custom_aliases import load
|
|
from tests.manage_fixture import bot_is_down, client_for, deps_at
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
RELOADED = {"status": "reloaded", "commands": 24}
|
|
|
|
|
|
def _bot_that_reloads(request: httpx.Request) -> httpx.Response:
|
|
assert request.method == "POST"
|
|
assert request.url.path == "/_reload"
|
|
return httpx.Response(200, json=RELOADED)
|
|
|
|
|
|
def test_the_table_starts_empty_and_a_put_fills_it(tmp_path: Path) -> None:
|
|
deps = deps_at(tmp_path, _bot_that_reloads)
|
|
with client_for(deps) as client:
|
|
assert client.get("/api/aliases").json() == {"aliases": {}}
|
|
answer = client.put("/api/aliases/cat", json={"triggers": ["kotek", "!kicia", "kotek"]})
|
|
listed = client.get("/api/aliases").json()
|
|
assert answer.status_code == 200
|
|
assert answer.json() == {"name": "cat", "triggers": ["!kotek", "!kicia"], **RELOADED}
|
|
assert listed == {"aliases": {"cat": ["!kotek", "!kicia"]}}
|
|
assert load(deps.aliases_path) == {"cat": ("!kotek", "!kicia")}
|
|
|
|
|
|
def test_an_empty_list_removes_the_command_from_the_file(tmp_path: Path) -> None:
|
|
deps = deps_at(tmp_path, _bot_that_reloads)
|
|
with client_for(deps) as client:
|
|
client.put("/api/aliases/cat", json={"triggers": ["!kotek"]})
|
|
client.put("/api/aliases/dog", json={"triggers": ["!piesek"]})
|
|
answer = client.put("/api/aliases/cat", json={"triggers": []})
|
|
assert answer.status_code == 200
|
|
assert load(deps.aliases_path) == {"dog": ("!piesek",)}
|
|
|
|
|
|
def test_a_trigger_of_several_words_is_refused_before_anything_is_written(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
deps = deps_at(tmp_path, _bot_that_reloads)
|
|
with client_for(deps) as client:
|
|
answer = client.put("/api/aliases/cat", json={"triggers": ["!kot ładny"]})
|
|
assert answer.status_code == 400
|
|
assert "one word" in answer.json()["detail"]
|
|
assert not deps.aliases_path.exists()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("code", "body", "expected"),
|
|
[
|
|
(409, {"status": "kept", "commands": 24, "reason": "trigger '!kot' is taken"}, "taken"),
|
|
(503, {"status": "connecting", "reason": "attempt 2: refused"}, "attempt 2"),
|
|
],
|
|
)
|
|
def test_a_reload_the_bot_refuses_puts_the_file_back(
|
|
tmp_path: Path, code: int, body: dict[str, object], expected: str
|
|
) -> None:
|
|
def bot(_request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(code, json=body)
|
|
|
|
deps = deps_at(tmp_path, bot)
|
|
before = json.dumps({"dog": ["!piesek"]})
|
|
deps.aliases_path.write_text(before, encoding="utf-8")
|
|
with client_for(deps) as client:
|
|
answer = client.put("/api/aliases/cat", json={"triggers": ["!kot"]})
|
|
assert answer.status_code == code
|
|
assert expected in answer.json()["detail"]
|
|
assert deps.aliases_path.read_text(encoding="utf-8") == before
|
|
|
|
|
|
def test_a_bot_that_does_not_answer_leaves_no_file_behind(tmp_path: Path) -> None:
|
|
deps = deps_at(tmp_path, bot_is_down)
|
|
with client_for(deps) as client:
|
|
answer = client.put("/api/aliases/cat", json={"triggers": ["!kotek"]})
|
|
assert answer.status_code == 502
|
|
assert not deps.aliases_path.exists()
|
|
|
|
|
|
def test_a_file_edited_by_hand_into_nonsense_is_reported_not_hidden(tmp_path: Path) -> None:
|
|
deps = deps_at(tmp_path, _bot_that_reloads)
|
|
deps.aliases_path.write_text("{oops", encoding="utf-8")
|
|
with client_for(deps) as client:
|
|
listed = client.get("/api/aliases")
|
|
put = client.put("/api/aliases/cat", json={"triggers": ["!kotek"]})
|
|
assert listed.status_code == 409
|
|
assert "cannot be read" in listed.json()["detail"]
|
|
assert put.status_code == 409
|
|
assert deps.aliases_path.read_text(encoding="utf-8") == "{oops"
|