mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-11 22:23:48 +01:00
Krzysztof kuhy Rudnicki
c35e382cd6
First live reply, as its own registered account. `!cat` and `!dog` from the
phone return images -- verified in the production shape (the bot *invited to
an already-existing group*) and under a systemd user unit.
Four failures, each of which left the bot looking perfectly healthy:
- Invited is not joined. Signal only adds an account outright if the adder
holds its profile key; with no profile the add becomes a pending invite,
and signal-cli cannot accept one addressed to a phone number ("Cannot find
service ID for self"). Group listed, websocket up, /healthz OK, zero
messages. register_bot.sh now sets the profile *before* the add, and
startup refuses on `member: false`.
- MODE=json-rpc-native is a GraalVM build with no AWT: text sends work,
every attachment 400s with "Can't load library: awt". Both mode guards now
demand exactly json-rpc, and start_container replaces a wrong-mode
container instead of reusing it.
- The API spells one group two ways -- /v1/groups returns `group.<base64>`,
envelopes carry what that decodes to -- so the scoping gate rejected the
bot's own group. e2e.sh now models both spellings; reverting the fix fails
4 scenarios, which is why the offline suite missed it before.
- BIND_PORT in .env was dead: __main__ read os.getenv before anything loaded
a .env. The bind address comes from Config now.
A taken port used to be uvicorn logging the collision and exiting 0, which
every supervisor reads as success. portcheck.py refuses to start, exits
non-zero, and names the holder by reading /proc directly -- no subprocess, no
PATH assumptions, testable against a fake proc tree. Default port is 8137:
not 8000, and below the ephemeral range so the kernel cannot take it.
The bot now runs as signal-bot.service (Restart=always, linger checked rather
than assumed, StartLimitBurst so a config error fails visibly instead of
looping). Verified by SIGKILL and by re-running a command under the unit.
100% branch coverage, 230 tests, all gates green.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013yaBT45KhHqxTKVCTA87iy
103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""The fake Signal server's own behaviour."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fake_signal.server import create_app, default_state
|
|
from fake_signal.state import PIXEL_PNG, State
|
|
|
|
|
|
@pytest.fixture
|
|
def state() -> State:
|
|
return State(base_url="http://testserver")
|
|
|
|
|
|
@pytest.fixture
|
|
def client(state: State) -> TestClient:
|
|
return TestClient(create_app(state))
|
|
|
|
|
|
def test_about_reports_a_websocket_mode(client: TestClient) -> None:
|
|
body = client.get("/v1/about").json()
|
|
assert body["mode"] == "json-rpc"
|
|
|
|
|
|
def test_qrcodelink_returns_a_png(client: TestClient) -> None:
|
|
"""The real endpoint returns an image, which is why the runbook opens it."""
|
|
response = client.get("/v1/qrcodelink", params={"device_name": "bot"})
|
|
assert response.headers["content-type"] == "image/png"
|
|
assert response.headers["X-Device-Name"] == "bot"
|
|
assert response.content == PIXEL_PNG
|
|
|
|
|
|
def test_send_is_captured_and_mints_attachment_ids(client: TestClient, state: State) -> None:
|
|
client.post("/v2/send", json={"message": "hi", "recipients": ["g"]})
|
|
client.post("/v2/send", json={"base64_attachments": ["A"], "recipients": ["g"]})
|
|
assert len(state.sent) == 2
|
|
assert state.attachments == ["attach-2-0"]
|
|
|
|
|
|
def test_attachments_can_be_listed_and_deleted(client: TestClient) -> None:
|
|
client.post("/v2/send", json={"base64_attachments": ["A"], "recipients": ["g"]})
|
|
listed = client.get("/v1/attachments").json()
|
|
assert listed == ["attach-1-0"]
|
|
assert client.delete(f"/v1/attachments/{listed[0]}").status_code == 204
|
|
assert client.get("/v1/attachments").json() == []
|
|
|
|
|
|
def test_deleting_an_unknown_attachment_is_a_noop(client: TestClient) -> None:
|
|
assert client.delete("/v1/attachments/nope").status_code == 204
|
|
|
|
|
|
def test_reset_clears_captured_state(client: TestClient, state: State) -> None:
|
|
client.post("/v2/send", json={"base64_attachments": ["A"], "recipients": ["g"]})
|
|
client.post("/_control/reset")
|
|
assert state.sent == []
|
|
assert state.attachments == []
|
|
|
|
|
|
def test_websocket_subscribes_and_receives_injected_frames(client: TestClient) -> None:
|
|
with client.websocket_connect("/v1/receive/+15550000000") as socket:
|
|
assert socket.receive_json()["envelope"]["subscribed"] == "+15550000000"
|
|
assert client.get("/_control/ready").json() == {"sockets": 1}
|
|
assert client.post("/_control/inject", json={"raw": "FRAME"}).json() == {"delivered": 1}
|
|
assert socket.receive_text() == "FRAME"
|
|
|
|
|
|
def test_socket_is_forgotten_after_disconnect(client: TestClient) -> None:
|
|
with client.websocket_connect("/v1/receive/+1") as socket:
|
|
socket.receive_json()
|
|
assert client.get("/_control/ready").json() == {"sockets": 0}
|
|
|
|
|
|
def test_inject_with_no_listeners_delivers_nothing(client: TestClient) -> None:
|
|
assert client.post("/_control/inject", json={"raw": "x"}).json() == {"delivered": 0}
|
|
|
|
|
|
def test_fixture_apis_mirror_the_real_response_shapes(client: TestClient) -> None:
|
|
cat = client.get("/_fixtures/cat-api").json()
|
|
dog = client.get("/_fixtures/dog-api").json()
|
|
assert cat[0]["url"].endswith("/_fixtures/image.png")
|
|
assert dog["message"].endswith("/_fixtures/image.png")
|
|
assert client.get("/_fixtures/image.png").content == PIXEL_PNG
|
|
|
|
|
|
def test_create_app_makes_its_own_state_when_none_given() -> None:
|
|
with TestClient(create_app()) as client:
|
|
assert client.get("/v1/about").status_code == 200
|
|
|
|
|
|
def test_default_state_reads_the_port_from_the_environment(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("FAKE_PORT", "9123")
|
|
assert default_state().base_url.endswith(":9123")
|
|
|
|
|
|
def test_control_sent_exposes_captured_payloads(client: TestClient) -> None:
|
|
client.post("/v2/send", json={"message": "hi", "recipients": ["R=="]})
|
|
assert client.get("/_control/sent").json() == [{"message": "hi", "recipients": ["R=="]}]
|