refactor: Notruf-Bridge nutzt seven-send-Library statt eigenem seven.io-Code

Die seven.io-Aufrufe (SMS + Voice/SSML) leben jetzt in der wiederverwendbaren
Library seven-send; die Bridge ist nur noch der HTTP-Adapter und ruft den
synchronen SevenClient via asyncio.to_thread. Eine getestete Stelle für die
API-Logik, ~50 Zeilen Duplikat raus.

- main.py: import SevenClient; _make_client() aus der Env; _dispatch() pro
  Kanal/Nummer; ohne API-Key sauberes "nicht gesetzt"-Ergebnis (kein Versand)
- Tests: mocken jetzt seven_send.client.httpx.post; neuer Test "ohne API-Key
  kein Versand" (6 Tests)
- Doku: Bridge-README + DEPLOYMENT 2.9 — seven-send als venv-Voraussetzung
  (pip install git+https://kitux.de/forgejo/dschlueter/seven_send.git)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-25 11:50:05 +02:00
commit ea0e3674ad
4 changed files with 71 additions and 82 deletions

View file

@ -1,4 +1,5 @@
import emergency_bridge.main as bridge
import seven_send.client as sc
from fastapi.testclient import TestClient
client = TestClient(bridge.app)
@ -12,29 +13,21 @@ PAYLOAD = {
}
def _fake_httpx(monkeypatch, capture):
class FakeResp:
status_code = 200
text = '{"success":"100","messages":[{"success":true}]}'
class FakeResp:
status_code = 200
text = '{"success":"100"}'
def json(self):
return {"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
def _fake_seven(monkeypatch, capture):
"""Mockt den seven.io-HTTP-Aufruf im seven_send-Client (eine Stelle für alles)."""
def fake_post(url, data=None, headers=None, timeout=None):
capture.append({"url": url, "data": data, "headers": headers})
return FakeResp()
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)
monkeypatch.setattr(sc.httpx, "post", fake_post)
def test_health():
@ -47,7 +40,7 @@ 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)
_fake_seven(monkeypatch, cap)
r = client.post("/notruf", json=PAYLOAD)
assert r.status_code == 200
@ -59,7 +52,7 @@ def test_notruf_sends_sms_and_voice(monkeypatch):
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)
# Anruf-Text wird von seven_send als SSML verpackt und XML-escaped
assert "<speak>" in voice["data"]["text"]
assert "&amp;" in voice["data"]["text"]
assert "&lt;Notfall&gt;" in voice["data"]["text"]
@ -74,8 +67,7 @@ def test_notruf_rejects_bad_token(monkeypatch):
def test_notruf_accepts_good_token(monkeypatch):
monkeypatch.setenv("BRIDGE_TOKEN", "secret")
monkeypatch.setenv("SEVENIO_API_KEY", "k")
cap = []
_fake_httpx(monkeypatch, cap)
_fake_seven(monkeypatch, [])
r = client.post("/notruf", json=PAYLOAD, headers={"Authorization": "Bearer secret"})
assert r.status_code == 200
@ -84,8 +76,23 @@ def test_notruf_respects_channels(monkeypatch):
monkeypatch.setenv("SEVENIO_API_KEY", "k")
monkeypatch.delenv("BRIDGE_TOKEN", raising=False)
cap = []
_fake_httpx(monkeypatch, cap)
_fake_seven(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")
def test_notruf_without_api_key(monkeypatch):
monkeypatch.delenv("SEVENIO_API_KEY", raising=False)
monkeypatch.delenv("BRIDGE_TOKEN", raising=False)
def boom(*a, **k): # ohne Key darf kein seven.io-Aufruf passieren
raise AssertionError("kein Versand ohne API-Key")
monkeypatch.setattr(sc.httpx, "post", boom)
r = client.post("/notruf", json=PAYLOAD)
assert r.status_code == 200
body = r.json()
assert body["any_sent"] is False
assert all("nicht gesetzt" in res["info"] for res in body["results"])