109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from services.shared.security import issue_app_token
|
|
from services.voice_adapter_service.app import app as voice_app
|
|
|
|
|
|
def _admin_headers() -> dict[str, str]:
|
|
return {"X-User": "admin", "X-Role": "admin"}
|
|
|
|
|
|
def test_voice_event_ingest_and_read_with_admin_headers():
|
|
client = TestClient(voice_app)
|
|
|
|
created = client.post(
|
|
"/integrations/voice/events",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"event_type": "call.started",
|
|
"call_id": "call_voice_test_admin",
|
|
"interaction_id": "int_voice_test_admin",
|
|
"payload": {"source": "unit-test"},
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
listed = client.get("/integrations/voice/events", headers=_admin_headers())
|
|
assert listed.status_code == 200
|
|
assert any(item["event_id"] == created.json()["event_id"] for item in listed.json())
|
|
|
|
|
|
def test_voice_event_ingest_allows_trusted_service_subject(monkeypatch):
|
|
monkeypatch.setenv("ALLOW_LEGACY_HEADER_AUTH", "0")
|
|
monkeypatch.setenv("APP_TOKEN_SECRET", "track9-voice-service-subject")
|
|
monkeypatch.setenv("VOICE_ADAPTER_TRUSTED_SERVICE_SUBJECTS", "svc:asterisk-bridge")
|
|
client = TestClient(voice_app)
|
|
|
|
token = issue_app_token(
|
|
subject="svc:asterisk-bridge",
|
|
username="asterisk-bridge",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="track9-test",
|
|
ttl_seconds=300,
|
|
)
|
|
allowed = client.post(
|
|
"/integrations/voice/events",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={
|
|
"event_type": "call.ended",
|
|
"call_id": "call_voice_test_service",
|
|
"interaction_id": "int_voice_test_service",
|
|
"payload": {"source": "service-test"},
|
|
},
|
|
)
|
|
assert allowed.status_code == 200
|
|
|
|
bad = issue_app_token(
|
|
subject="svc:other",
|
|
username="other-service",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="track9-test",
|
|
ttl_seconds=300,
|
|
)
|
|
denied = client.post(
|
|
"/integrations/voice/events",
|
|
headers={"Authorization": f"Bearer {bad}"},
|
|
json={
|
|
"event_type": "call.ended",
|
|
"call_id": "call_voice_test_denied",
|
|
"interaction_id": "int_voice_test_denied",
|
|
"payload": {"source": "service-test"},
|
|
},
|
|
)
|
|
assert denied.status_code == 403
|
|
|
|
|
|
def test_voice_event_ingest_is_idempotent_by_source_event_id():
|
|
client = TestClient(voice_app)
|
|
source_event_id = "bridge_evt_voice_idempotent"
|
|
payload = {
|
|
"event_type": "call.started",
|
|
"call_id": "call_voice_idempotent",
|
|
"interaction_id": "int_voice_idempotent",
|
|
"source_event_id": source_event_id,
|
|
"payload": {"source": "bridge-test"},
|
|
}
|
|
|
|
first = client.post(
|
|
"/integrations/voice/events",
|
|
headers=_admin_headers(),
|
|
json=payload,
|
|
)
|
|
second = client.post(
|
|
"/integrations/voice/events",
|
|
headers=_admin_headers(),
|
|
json=payload,
|
|
)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert first.json()["event_id"] == second.json()["event_id"]
|
|
assert first.json()["source_event_id"] == source_event_id
|
|
|
|
listed = client.get("/integrations/voice/events", headers=_admin_headers())
|
|
assert listed.status_code == 200
|
|
matching = [item for item in listed.json() if item["call_id"] == payload["call_id"]]
|
|
assert len(matching) == 1
|