511 lines
19 KiB
Python
511 lines
19 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from services.interaction_service.app import app as interaction_app
|
|
from services.ivr_service.app import app as ivr_app
|
|
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 _supervisor_headers() -> dict[str, str]:
|
|
return {"X-User": "supervisor", "X-Role": "supervisor"}
|
|
|
|
|
|
def _valid_flow(queue_id: str, resolved_queue_id: str) -> dict:
|
|
return {
|
|
"name": "Queue IVR",
|
|
"description": "DTMF flow",
|
|
"queue_id": queue_id,
|
|
"entry_node_id": "root",
|
|
"flow_json": {
|
|
"nodes": [
|
|
{
|
|
"node_id": "root",
|
|
"prompt_text": "Press 1 for support",
|
|
"prompt_audio_key": "ivr/demo-root",
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "root",
|
|
"options": [
|
|
{"digit": "1", "target_node_id": "support"},
|
|
],
|
|
},
|
|
{
|
|
"node_id": "support",
|
|
"prompt_text": "Routing to support",
|
|
"prompt_audio_key": "ivr/demo-support",
|
|
"is_terminal": True,
|
|
"outcome_code": "support_route",
|
|
"resolved_queue_id": resolved_queue_id,
|
|
"resolved_queue_code": "support",
|
|
"options": [],
|
|
},
|
|
]
|
|
},
|
|
"is_active": True,
|
|
}
|
|
|
|
|
|
def _trusted_service_headers(subject: str = "svc:asterisk-bridge") -> dict[str, str]:
|
|
token = issue_app_token(
|
|
subject=subject,
|
|
username="asterisk-bridge",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="asterisk-bridge",
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_create_validate_and_activate_flows():
|
|
client = TestClient(ivr_app)
|
|
queue_id = "ivr_queue_validation"
|
|
|
|
invalid = client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"name": "Broken Flow",
|
|
"description": "",
|
|
"queue_id": queue_id,
|
|
"entry_node_id": "root",
|
|
"flow_json": {"nodes": [{"node_id": "root", "is_terminal": False, "options": []}]},
|
|
"is_active": True,
|
|
},
|
|
)
|
|
assert invalid.status_code == 400
|
|
|
|
first = client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json=_valid_flow(queue_id, "support_queue_a"),
|
|
)
|
|
assert first.status_code == 200
|
|
first_flow_id = first.json()["flow_id"]
|
|
assert first.json()["is_active"] is True
|
|
|
|
second_payload = _valid_flow(queue_id, "support_queue_b")
|
|
second_payload["name"] = "Replacement Flow"
|
|
second = client.post("/ivr/flows", headers=_admin_headers(), json=second_payload)
|
|
assert second.status_code == 200
|
|
second_flow_id = second.json()["flow_id"]
|
|
assert second_flow_id != first_flow_id
|
|
|
|
first_detail = client.get(f"/ivr/flows/{first_flow_id}", headers=_supervisor_headers())
|
|
assert first_detail.status_code == 200
|
|
assert first_detail.json()["is_active"] is False
|
|
|
|
activated = client.post(f"/ivr/flows/{first_flow_id}/activate", headers=_admin_headers())
|
|
assert activated.status_code == 200
|
|
assert activated.json()["is_active"] is True
|
|
|
|
second_detail = client.get(f"/ivr/flows/{second_flow_id}", headers=_supervisor_headers())
|
|
assert second_detail.status_code == 200
|
|
assert second_detail.json()["is_active"] is False
|
|
|
|
missing_queue_code = client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"name": "Broken Terminal",
|
|
"description": "",
|
|
"queue_id": queue_id,
|
|
"entry_node_id": "root",
|
|
"flow_json": {
|
|
"nodes": [
|
|
{
|
|
"node_id": "root",
|
|
"prompt_text": "Press 1",
|
|
"is_terminal": False,
|
|
"options": [{"digit": "1", "target_node_id": "terminal"}],
|
|
},
|
|
{
|
|
"node_id": "terminal",
|
|
"prompt_text": "Done",
|
|
"is_terminal": True,
|
|
"outcome_code": "done",
|
|
"resolved_queue_id": "support_queue_a",
|
|
"options": [],
|
|
},
|
|
]
|
|
},
|
|
"is_active": False,
|
|
},
|
|
)
|
|
assert missing_queue_code.status_code == 400
|
|
assert "resolved_queue_code" in missing_queue_code.json()["detail"]
|
|
|
|
|
|
def test_flow_validation_accepts_prompt_sequence_without_single_prompt_key():
|
|
client = TestClient(ivr_app)
|
|
|
|
response = client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"name": "Prompt Sequence Flow",
|
|
"description": "Mixed language prompt split into segments",
|
|
"queue_id": "ivr_prompt_sequence_queue",
|
|
"entry_node_id": "root",
|
|
"flow_json": {
|
|
"nodes": [
|
|
{
|
|
"node_id": "root",
|
|
"prompt_text": "Combined bilingual prompt",
|
|
"prompt_sequence": [
|
|
{
|
|
"prompt_audio_key": "ivr/demo-language-ru",
|
|
"prompt_text": "Russian segment",
|
|
"language": "ru",
|
|
},
|
|
{
|
|
"prompt_audio_key": "ivr/demo-language-kz",
|
|
"prompt_text": "Kazakh segment",
|
|
"language": "kz",
|
|
},
|
|
],
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "root",
|
|
"options": [
|
|
{"digit": "1", "target_node_id": "terminal"},
|
|
],
|
|
},
|
|
{
|
|
"node_id": "terminal",
|
|
"prompt_text": "Done",
|
|
"is_terminal": True,
|
|
"outcome_code": "done",
|
|
"resolved_queue_id": "support_queue_a",
|
|
"resolved_queue_code": "support",
|
|
"options": [],
|
|
},
|
|
]
|
|
},
|
|
"is_active": False,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["flow_json"]["nodes"][0]["prompt_sequence"][0]["prompt_audio_key"] == "ivr/demo-language-ru"
|
|
assert payload["flow_json"]["nodes"][0]["prompt_sequence"][1]["prompt_audio_key"] == "ivr/demo-language-kz"
|
|
|
|
|
|
def test_ivr_session_runtime_emits_voice_event_and_timeline():
|
|
ivr_client = TestClient(ivr_app)
|
|
voice_client = TestClient(voice_app)
|
|
interaction_client = TestClient(interaction_app)
|
|
|
|
interaction = interaction_client.post(
|
|
"/interactions",
|
|
json={
|
|
"channel": "voice",
|
|
"subject": "IVR runtime trace",
|
|
"queue_id": "ivr_runtime_queue",
|
|
"priority": 3,
|
|
},
|
|
headers={"X-User": "operator", "X-Role": "operator"},
|
|
)
|
|
assert interaction.status_code == 200
|
|
interaction_id = interaction.json()["interaction_id"]
|
|
|
|
flow = ivr_client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json=_valid_flow("ivr_runtime_queue", "ivr_resolved_support"),
|
|
)
|
|
assert flow.status_code == 200
|
|
flow_id = flow.json()["flow_id"]
|
|
|
|
started = ivr_client.post(
|
|
"/ivr/sessions/start",
|
|
headers=_supervisor_headers(),
|
|
json={
|
|
"call_id": "call_ivr_runtime",
|
|
"queue_id": "ivr_runtime_queue",
|
|
"interaction_id": interaction_id,
|
|
},
|
|
)
|
|
assert started.status_code == 200
|
|
assert started.json()["session"]["flow_id"] == flow_id
|
|
session_id = started.json()["session"]["session_id"]
|
|
|
|
stepped = ivr_client.post(
|
|
f"/ivr/sessions/{session_id}/dtmf",
|
|
headers=_supervisor_headers(),
|
|
json={"digit": "1"},
|
|
)
|
|
assert stepped.status_code == 200
|
|
assert stepped.json()["completed"] is True
|
|
assert stepped.json()["session"]["status"] == "completed"
|
|
assert stepped.json()["session"]["outcome_code"] == "support_route"
|
|
assert stepped.json()["session"]["resolved_queue_code"] == "support"
|
|
|
|
repeated = ivr_client.post(
|
|
f"/ivr/sessions/{session_id}/dtmf",
|
|
headers=_supervisor_headers(),
|
|
json={"digit": "1"},
|
|
)
|
|
assert repeated.status_code == 409
|
|
|
|
voice_events = voice_client.get("/integrations/voice/events", headers=_admin_headers())
|
|
assert voice_events.status_code == 200
|
|
assert any(
|
|
item["event_type"] == "ivr.completed"
|
|
and item["payload"]["session_id"] == session_id
|
|
and item["payload"]["flow_id"] == flow_id
|
|
for item in voice_events.json()
|
|
)
|
|
|
|
timeline = interaction_client.get(f"/interactions/{interaction_id}/timeline")
|
|
assert timeline.status_code == 200
|
|
actions = [item["action"] for item in timeline.json()["events"]]
|
|
assert "ivr.session.started" in actions
|
|
assert "ivr.step.completed" in actions
|
|
assert "ivr.completed" in actions
|
|
|
|
|
|
def test_no_input_runtime_path_supports_trusted_service_actor():
|
|
ivr_client = TestClient(ivr_app)
|
|
interaction_client = TestClient(interaction_app)
|
|
|
|
interaction = interaction_client.post(
|
|
"/interactions",
|
|
json={
|
|
"channel": "voice",
|
|
"subject": "IVR no input trace",
|
|
"queue_id": "ivr_no_input_queue",
|
|
"priority": 3,
|
|
},
|
|
headers={"X-User": "operator", "X-Role": "operator"},
|
|
)
|
|
assert interaction.status_code == 200
|
|
interaction_id = interaction.json()["interaction_id"]
|
|
|
|
flow = ivr_client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"name": "No Input Flow",
|
|
"description": "No input route",
|
|
"queue_id": "ivr_no_input_queue",
|
|
"entry_node_id": "root",
|
|
"flow_json": {
|
|
"nodes": [
|
|
{
|
|
"node_id": "root",
|
|
"prompt_text": "Press 1 for support",
|
|
"prompt_audio_key": "ivr/no-input-root",
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "root",
|
|
"no_input_target_node_id": "timeout",
|
|
"options": [{"digit": "1", "target_node_id": "support"}],
|
|
},
|
|
{
|
|
"node_id": "support",
|
|
"prompt_text": "Support queue",
|
|
"prompt_audio_key": "ivr/no-input-support",
|
|
"is_terminal": True,
|
|
"outcome_code": "support_route",
|
|
"resolved_queue_id": "ivr_support_queue",
|
|
"resolved_queue_code": "support",
|
|
"options": [],
|
|
},
|
|
{
|
|
"node_id": "timeout",
|
|
"prompt_text": "Routing to default queue",
|
|
"prompt_audio_key": "ivr/no-input-timeout",
|
|
"is_terminal": True,
|
|
"outcome_code": "timeout_route",
|
|
"resolved_queue_id": "ivr_timeout_queue",
|
|
"resolved_queue_code": "voice_lab",
|
|
"options": [],
|
|
},
|
|
]
|
|
},
|
|
"is_active": True,
|
|
},
|
|
)
|
|
assert flow.status_code == 200
|
|
|
|
started = ivr_client.post(
|
|
"/ivr/sessions/start",
|
|
headers=_trusted_service_headers(),
|
|
json={
|
|
"call_id": "call_ivr_no_input",
|
|
"queue_id": "ivr_no_input_queue",
|
|
"interaction_id": interaction_id,
|
|
},
|
|
)
|
|
assert started.status_code == 200
|
|
session_id = started.json()["session"]["session_id"]
|
|
|
|
no_input = ivr_client.post(
|
|
f"/ivr/sessions/{session_id}/no-input",
|
|
headers=_trusted_service_headers(),
|
|
)
|
|
assert no_input.status_code == 200
|
|
assert no_input.json()["completed"] is True
|
|
assert no_input.json()["session"]["status"] == "completed"
|
|
assert no_input.json()["session"]["resolved_queue_id"] == "ivr_timeout_queue"
|
|
assert no_input.json()["session"]["resolved_queue_code"] == "voice_lab"
|
|
|
|
blocked = ivr_client.post(
|
|
"/ivr/sessions/start",
|
|
headers=_trusted_service_headers("svc:not-allowed"),
|
|
json={"call_id": "call_blocked", "queue_id": "ivr_no_input_queue"},
|
|
)
|
|
assert blocked.status_code == 403
|
|
|
|
timeline = interaction_client.get(f"/interactions/{interaction_id}/timeline")
|
|
assert timeline.status_code == 200
|
|
actions = [item["action"] for item in timeline.json()["events"]]
|
|
assert "ivr.step.no_input" in actions
|
|
assert "ivr.completed" in actions
|
|
|
|
|
|
def test_multistep_language_selection_flow_routes_after_second_digit():
|
|
ivr_client = TestClient(ivr_app)
|
|
|
|
flow = ivr_client.post(
|
|
"/ivr/flows",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"name": "Language IVR",
|
|
"description": "Greeting and language choice before routing",
|
|
"queue_id": "ivr_language_queue",
|
|
"entry_node_id": "root",
|
|
"flow_json": {
|
|
"nodes": [
|
|
{
|
|
"node_id": "root",
|
|
"prompt_text": "Welcome. Press 1 for Russian. Press 2 for Kazakh.",
|
|
"prompt_audio_key": "ivr/language-root",
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "root",
|
|
"no_input_target_node_id": "root",
|
|
"options": [
|
|
{"digit": "1", "target_node_id": "menu_ru"},
|
|
{"digit": "2", "target_node_id": "menu_kz"},
|
|
],
|
|
},
|
|
{
|
|
"node_id": "menu_ru",
|
|
"prompt_text": "Russian menu",
|
|
"prompt_audio_key": "ivr/language-menu-ru",
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "menu_ru",
|
|
"no_input_target_node_id": "menu_ru",
|
|
"options": [
|
|
{"digit": "1", "target_node_id": "sales_ru"},
|
|
{"digit": "2", "target_node_id": "support_ru"},
|
|
],
|
|
},
|
|
{
|
|
"node_id": "menu_kz",
|
|
"prompt_text": "Kazakh menu",
|
|
"prompt_audio_key": "ivr/language-menu-kz",
|
|
"is_terminal": False,
|
|
"invalid_target_node_id": "menu_kz",
|
|
"no_input_target_node_id": "menu_kz",
|
|
"options": [
|
|
{"digit": "1", "target_node_id": "sales_kz"},
|
|
{"digit": "2", "target_node_id": "support_kz"},
|
|
],
|
|
},
|
|
{
|
|
"node_id": "sales_ru",
|
|
"prompt_text": "Routing to sales",
|
|
"prompt_audio_key": "ivr/language-sales-ru",
|
|
"is_terminal": True,
|
|
"outcome_code": "sales_route_ru",
|
|
"resolved_queue_id": "ivr_sales_queue",
|
|
"resolved_queue_code": "ivr_sales_ai_ru",
|
|
"options": [],
|
|
},
|
|
{
|
|
"node_id": "support_ru",
|
|
"prompt_text": "Routing to support",
|
|
"prompt_audio_key": "ivr/language-support-ru",
|
|
"is_terminal": True,
|
|
"outcome_code": "support_route_ru",
|
|
"resolved_queue_id": "ivr_support_queue",
|
|
"resolved_queue_code": "ivr_support_ai_ru",
|
|
"options": [],
|
|
},
|
|
{
|
|
"node_id": "sales_kz",
|
|
"prompt_text": "Routing to sales",
|
|
"prompt_audio_key": "ivr/language-sales-kz",
|
|
"is_terminal": True,
|
|
"outcome_code": "sales_route_kz",
|
|
"resolved_queue_id": "ivr_sales_queue",
|
|
"resolved_queue_code": "ivr_sales_ai_kz",
|
|
"options": [],
|
|
},
|
|
{
|
|
"node_id": "support_kz",
|
|
"prompt_text": "Routing to support",
|
|
"prompt_audio_key": "ivr/language-support-kz",
|
|
"is_terminal": True,
|
|
"outcome_code": "support_route_kz",
|
|
"resolved_queue_id": "ivr_support_queue",
|
|
"resolved_queue_code": "ivr_support_ai_kz",
|
|
"options": [],
|
|
},
|
|
]
|
|
},
|
|
"is_active": True,
|
|
},
|
|
)
|
|
assert flow.status_code == 200
|
|
|
|
started = ivr_client.post(
|
|
"/ivr/sessions/start",
|
|
headers=_trusted_service_headers(),
|
|
json={
|
|
"call_id": "call_language_ivr",
|
|
"queue_id": "ivr_language_queue",
|
|
},
|
|
)
|
|
assert started.status_code == 200
|
|
assert started.json()["completed"] is False
|
|
assert started.json()["current_node"]["node_id"] == "root"
|
|
session_id = started.json()["session"]["session_id"]
|
|
|
|
select_language = ivr_client.post(
|
|
f"/ivr/sessions/{session_id}/dtmf",
|
|
headers=_trusted_service_headers(),
|
|
json={"digit": "2"},
|
|
)
|
|
assert select_language.status_code == 200
|
|
assert select_language.json()["completed"] is False
|
|
assert select_language.json()["session"]["current_node_id"] == "menu_kz"
|
|
assert select_language.json()["session"]["entered_digits"] == ["2"]
|
|
|
|
choose_support = ivr_client.post(
|
|
f"/ivr/sessions/{session_id}/dtmf",
|
|
headers=_trusted_service_headers(),
|
|
json={"digit": "2"},
|
|
)
|
|
assert choose_support.status_code == 200
|
|
assert choose_support.json()["completed"] is True
|
|
assert choose_support.json()["session"]["status"] == "completed"
|
|
assert choose_support.json()["session"]["entered_digits"] == ["2", "2"]
|
|
assert choose_support.json()["session"]["resolved_queue_id"] == "ivr_support_queue"
|
|
assert choose_support.json()["session"]["resolved_queue_code"] == "ivr_support_ai_kz"
|
|
assert choose_support.json()["session"]["outcome_code"] == "support_route_kz"
|
|
|
|
|
|
def test_start_without_active_flow_returns_404():
|
|
client = TestClient(ivr_app)
|
|
|
|
response = client.post(
|
|
"/ivr/sessions/start",
|
|
headers=_supervisor_headers(),
|
|
json={"call_id": "call_missing_flow", "queue_id": "missing_queue"},
|
|
)
|
|
assert response.status_code == 404
|