feat: Notruf-Bridge (seven.io) — Webhook-Empfänger für SMS + Anruf

Konkreter Provider hinter dem provider-agnostischen EMERGENCY_WEBHOOK_URL:
ein kleiner FastAPI-Dienst (emergency_bridge/), der das Notruf-Payload der
App entgegennimmt und pro Telefonnummer SMS und/oder TTS-Anruf über die
seven.io-API auslöst.

- main.py: POST /notruf (Bearer-Token-Prüfung), GET /health; seven.io
  SMS (/api/sms) + Voice (/api/voice, Text als SSML, XML-escaped), best
  effort mit Statuscode-100-Auswertung; Kanäle/Texte kommen aus dem Payload
- Betrieb: eigener systemd-Service auf 127.0.0.1:8090, läuft in der
  vorhandenen venv; .env.example + README mit Installations- und
  Verdrahtungsschritten
- Tests: SMS+Anruf-Fan-out, SSML-Escaping, Token-Auth, Kanal-Filter (5 Tests)
- DEPLOYMENT.md: optionaler Abschnitt 2.9 mit Verweis auf die Bridge

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-25 10:53:34 +02:00
commit 7addb903b2
7 changed files with 353 additions and 0 deletions

View file

@ -0,0 +1,91 @@
import emergency_bridge.main as bridge
from fastapi.testclient import TestClient
client = TestClient(bridge.app)
PAYLOAD = {
"event": "emergency",
"category": "manual",
"channels": ["sms", "call"],
"phones": ["+4915112345678"],
"message": {"sms": "NOTRUF: Test", "call": "Achtung. Test & <Notfall>."},
}
def _fake_httpx(monkeypatch, capture):
class FakeResp:
status_code = 200
text = '{"success":"100","messages":[{"success":true}]}'
def json(self):
return {"success": "100", "messages": [{"success": True}]}
class FakeClient:
def __init__(self, *a, **k):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def post(self, url, data=None, headers=None):
capture.append({"url": url, "data": data, "headers": headers})
return FakeResp()
monkeypatch.setattr(bridge.httpx, "AsyncClient", FakeClient)
def test_health():
r = client.get("/health")
assert r.status_code == 200
assert r.json()["status"] == "ok"
def test_notruf_sends_sms_and_voice(monkeypatch):
monkeypatch.setenv("SEVENIO_API_KEY", "k")
monkeypatch.delenv("BRIDGE_TOKEN", raising=False)
cap = []
_fake_httpx(monkeypatch, cap)
r = client.post("/notruf", json=PAYLOAD)
assert r.status_code == 200
assert r.json()["any_sent"] is True
assert len(cap) == 2 # eine SMS + ein Anruf
sms = next(c for c in cap if c["url"].endswith("/sms"))
voice = next(c for c in cap if c["url"].endswith("/voice"))
assert sms["headers"]["X-Api-Key"] == "k"
assert sms["data"]["to"] == "+4915112345678"
assert sms["data"]["text"] == "NOTRUF: Test"
# Anruf-Text als SSML, XML-escaped (& und <> dürfen das SSML nicht zerlegen)
assert "<speak>" in voice["data"]["text"]
assert "&amp;" in voice["data"]["text"]
assert "&lt;Notfall&gt;" in voice["data"]["text"]
def test_notruf_rejects_bad_token(monkeypatch):
monkeypatch.setenv("BRIDGE_TOKEN", "secret")
r = client.post("/notruf", json=PAYLOAD, headers={"Authorization": "Bearer wrong"})
assert r.status_code == 401
def test_notruf_accepts_good_token(monkeypatch):
monkeypatch.setenv("BRIDGE_TOKEN", "secret")
monkeypatch.setenv("SEVENIO_API_KEY", "k")
cap = []
_fake_httpx(monkeypatch, cap)
r = client.post("/notruf", json=PAYLOAD, headers={"Authorization": "Bearer secret"})
assert r.status_code == 200
def test_notruf_respects_channels(monkeypatch):
monkeypatch.setenv("SEVENIO_API_KEY", "k")
monkeypatch.delenv("BRIDGE_TOKEN", raising=False)
cap = []
_fake_httpx(monkeypatch, cap)
r = client.post("/notruf", json={**PAYLOAD, "channels": ["sms"]})
assert r.status_code == 200
assert len(cap) == 1
assert cap[0]["url"].endswith("/sms")