from fastapi.testclient import TestClient from services.customer_service.app import app as customer_app from services.interaction_service.app import app as interaction_app from services.ivr_service.app import app as ivr_app from services.routing_service.app import app as routing_app from services.shared.db import get_session from services.shared.sql_models import ( AsteriskCallLinkRow, CallRecordingRow, Customer, CustomerExternalIdentity, Queue, TelegramMessageRow, TelegramThreadRow, VoiceAISessionRow, VoiceTranscriptSegmentRow, ) def test_customer_and_interaction_lifecycle(): customer_client = TestClient(customer_app) interaction_client = TestClient(interaction_app) customer = customer_client.post( "/customers", json={"display_name": "Test Client", "phones": ["+77010000001"], "preferred_phone": "+77010000001"}, ) assert customer.status_code == 200 customer_id = customer.json()["customer_id"] interaction = interaction_client.post( "/interactions", json={ "channel": "voice", "subject": "Need support", "customer_id": customer_id, "queue_id": "q_main", "priority": 2, }, headers={"X-User": "operator", "X-Role": "operator"}, ) assert interaction.status_code == 200 interaction_id = interaction.json()["interaction_id"] assigned = interaction_client.patch( f"/interactions/{interaction_id}/assign", json={"assignee": "operator_a"}, headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert assigned.status_code == 200 assert assigned.json()["status"] == "in_progress" escalated = interaction_client.post( f"/interactions/{interaction_id}/escalate", json={"target_queue_id": "line2"}, headers={"X-User": "operator", "X-Role": "operator"}, ) assert escalated.status_code == 200 assert escalated.json()["status"] == "escalated" closed = interaction_client.patch( f"/interactions/{interaction_id}/status", json={"status": "closed"}, headers={"X-User": "operator", "X-Role": "operator"}, ) assert closed.status_code == 200 assert closed.json()["status"] == "closed" timeline = interaction_client.get(f"/interactions/{interaction_id}/timeline") assert timeline.status_code == 200 assert len(timeline.json()["events"]) >= 4 def test_customer_history_aggregates_telegram_and_voice_context(): customer_client = TestClient(customer_app) interaction_client = TestClient(interaction_app) customer = customer_client.post( "/customers", json={ "display_name": "History Client", "phones": ["+77010000019"], "preferred_phone": "+77010000019", "tags": ["priority"], }, ) assert customer.status_code == 200 customer_id = customer.json()["customer_id"] voice_interaction = interaction_client.post( "/interactions", json={ "channel": "voice", "subject": "Need callback after AI handoff", "customer_id": customer_id, "queue_id": "q_voice", "priority": 2, }, headers={"X-User": "operator", "X-Role": "operator"}, ) assert voice_interaction.status_code == 200 voice_interaction_id = voice_interaction.json()["interaction_id"] telegram_interaction = interaction_client.post( "/interactions", json={ "channel": "telegram", "subject": "Order issue in Telegram", "customer_id": customer_id, "queue_id": "q_chat", "priority": 3, }, headers={"X-User": "operator", "X-Role": "operator"}, ) assert telegram_interaction.status_code == 200 telegram_interaction_id = telegram_interaction.json()["interaction_id"] assigned = interaction_client.patch( f"/interactions/{telegram_interaction_id}/assign", json={"assignee": "operator_a"}, headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert assigned.status_code == 200 session = get_session() try: session.add( TelegramThreadRow( thread_id="tgt_history_customer", chat_id="chat_history_customer", interaction_id=telegram_interaction_id, telegram_user_id="tg_hist_1", username="history_client", display_name="History Client", queue_id="q_chat", status="in_progress", claimed_by_user="operator_a", claimed_at="2026-04-03T09:05:00+00:00", ai_session_id="ais_hist_1", ai_state="human_owned", ai_handoff_reason="Нужно проверить статус заказа вручную", ai_last_model_at="2026-04-03T09:04:00+00:00", last_message_at="2026-04-03T09:06:00+00:00", last_message_preview="Клиент просит уточнить статус заказа", created_at="2026-04-03T09:00:00+00:00", updated_at="2026-04-03T09:06:00+00:00", ) ) session.add( TelegramMessageRow( message_id="tgm_history_customer_in", thread_id="tgt_history_customer", interaction_id=telegram_interaction_id, chat_id="chat_history_customer", text="Клиент просит уточнить статус заказа", customer_external_id="telegram:history_client", direction="inbound", telegram_message_id_external="501", operator_user=None, author_type="customer", author_id="tg_hist_1", delivery_status=None, payload_json="{}", created_at="2026-04-03T09:06:00+00:00", ) ) session.add( AsteriskCallLinkRow( call_id="call_history_customer", linked_id="linked_history_customer", queue_code="voice_support", queue_id="q_voice", interaction_id=voice_interaction_id, caller_number="+77010000019", caller_name="History Client", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at="2026-04-03T09:11:00+00:00", operator_extension="2001", channel_name="PJSIP/2001-000001", voice_session_id="vas_history_1", ai_session_id="ais_voice_1", ai_state="handoff_required", ai_handoff_reason="AI не смог подтвердить оплату по заказу", ai_last_model_at="2026-04-03T09:12:00+00:00", started_at="2026-04-03T09:10:00+00:00", connected_at="2026-04-03T09:11:00+00:00", ended_at=None, updated_at="2026-04-03T09:12:00+00:00", ) ) session.add( CallRecordingRow( recording_id="rec_history_customer", channel="voice", call_id="call_history_customer", interaction_id=voice_interaction_id, source_event_id="evt_history_customer", file_name="history-call.wav", storage_backend="local_fs", storage_path="history/history-call.wav", mime_type="audio/wav", size_bytes=1024, duration_seconds=48, checksum_sha256="abc123", status="ready", recorded_at="2026-04-03T09:13:00+00:00", created_at="2026-04-03T09:13:00+00:00", updated_at="2026-04-03T09:13:00+00:00", archived_at=None, ) ) session.add( VoiceTranscriptSegmentRow( segment_id="seg_history_customer", session_id="vas_history_1", call_id="call_history_customer", interaction_id=voice_interaction_id, speaker="caller", source_type="asr", sequence_no=1, text="Мне нужно, чтобы оператор подтвердил оплату по счёту", confidence=0.93, is_final=True, barge_in_interrupted=False, payload_json="{}", created_at="2026-04-03T09:11:30+00:00", ) ) session.commit() finally: session.close() history = customer_client.get(f"/customers/{customer_id}/history") assert history.status_code == 200 payload = history.json() assert payload["customer"]["customer_id"] == customer_id assert payload["summary"]["open_cases"] == 2 assert payload["summary"]["primary_phone"] == "+77010000019" assert payload["summary"]["primary_telegram_thread_id"] == "tgt_history_customer" assert set(payload["summary"]["active_channels"]) == {"telegram", "voice"} assert {item["interaction_id"] for item in payload["interactions"]} == { voice_interaction_id, telegram_interaction_id, } assert payload["telegram_threads"][0]["thread_id"] == "tgt_history_customer" assert payload["live_calls"][0]["call_id"] == "call_history_customer" assert payload["live_calls"][0]["has_recording"] is True titles = [item["title"] for item in payload["history"]] assert "Telegram диалог" in titles assert "Обращение назначено" in titles assert any(title.startswith("Звонок •") for title in titles) assert any(item["body"] == "Клиент просит уточнить статус заказа" for item in payload["history"]) assert any("оператор" in (item["note"] or "") for item in payload["history"]) def test_operator_can_fix_customer_name_and_sync_voice_context(): customer_client = TestClient(customer_app) interaction_client = TestClient(interaction_app) customer = customer_client.post( "/customers", json={ "display_name": "Номер 1001", "phones": ["+77010001001"], "preferred_phone": "+77010001001", }, ) assert customer.status_code == 200 customer_id = customer.json()["customer_id"] interaction = interaction_client.post( "/interactions", json={ "channel": "voice", "subject": "Inbound call 1001", "customer_id": customer_id, "queue_id": "q_voice", "priority": 2, }, headers={"X-User": "operator", "X-Role": "operator"}, ) assert interaction.status_code == 200 interaction_id = interaction.json()["interaction_id"] session = get_session() try: session.add( CustomerExternalIdentity( identity_id="cei_voice_fix_name", customer_id=customer_id, channel="voice", external_subject="+77010001001", display_name_snapshot="Номер 1001", created_at="2026-04-05T09:00:00+00:00", updated_at="2026-04-05T09:00:00+00:00", ) ) session.add( AsteriskCallLinkRow( call_id="call_fix_name_customer", linked_id="linked_fix_name_customer", queue_code="voice_support", queue_id="q_voice", interaction_id=interaction_id, caller_number="+77010001001", caller_name="Номер 1001", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at="2026-04-05T09:01:00+00:00", operator_extension="2001", channel_name="PJSIP/2001-000010", voice_session_id="vas_fix_name_customer", ai_session_id="ais_fix_name_customer", ai_state="human_owned", ai_handoff_reason="Оператор уточняет имя клиента", ai_last_model_at="2026-04-05T09:02:00+00:00", voice_start_language="ru", customer_name_status="name_followup_required", customer_name_value="Айдос?", customer_name_source="voice_followup", customer_name_resolved_at="2026-04-05T09:02:00+00:00", started_at="2026-04-05T09:00:00+00:00", connected_at="2026-04-05T09:01:00+00:00", ended_at=None, updated_at="2026-04-05T09:02:00+00:00", ) ) session.add( VoiceAISessionRow( session_id="vas_fix_name_customer", call_id="call_fix_name_customer", linked_id="linked_fix_name_customer", interaction_id=interaction_id, customer_id=customer_id, queue_id="q_voice", ai_session_id="ais_fix_name_customer", agent_profile="voice_support", language="ru", asr_provider="mock", tts_provider="mock", status="handoff_requested", handoff_reason="Оператор уточняет имя клиента", handoff_target_queue_id="q_voice", voice_start_language="ru", customer_name_status="name_followup_required", customer_name_value="Айдос?", customer_name_source="voice_followup", customer_name_resolved_at="2026-04-05T09:02:00+00:00", media_uuid=None, media_status=None, media_connected_at=None, media_ended_at=None, last_media_frame_at=None, disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at="2026-04-05T09:00:00+00:00", updated_at="2026-04-05T09:02:00+00:00", ended_at=None, ) ) session.commit() finally: session.close() updated = customer_client.patch( f"/customers/{customer_id}", json={"display_name": "Айдос", "source": "manual"}, ) assert updated.status_code == 200 assert updated.json()["display_name"] == "Айдос" session = get_session() try: customer_row = session.query(Customer).filter(Customer.customer_id == customer_id).one() identity_row = session.query(CustomerExternalIdentity).filter(CustomerExternalIdentity.customer_id == customer_id).one() call_row = session.query(AsteriskCallLinkRow).filter(AsteriskCallLinkRow.call_id == "call_fix_name_customer").one() voice_row = session.query(VoiceAISessionRow).filter(VoiceAISessionRow.call_id == "call_fix_name_customer").one() assert customer_row.display_name == "Айдос" assert identity_row.display_name_snapshot == "Айдос" assert call_row.caller_name == "Айдос" assert call_row.customer_name_status == "name_obtained" assert call_row.customer_name_value == "Айдос" assert call_row.customer_name_source == "manual" assert voice_row.customer_name_status == "name_obtained" assert voice_row.customer_name_value == "Айдос" assert voice_row.customer_name_source == "manual" finally: session.close() history = customer_client.get(f"/customers/{customer_id}/history") assert history.status_code == 200 assert history.json()["customer"]["display_name"] == "Айдос" assert history.json()["live_calls"][0]["caller_name"] == "Айдос" def test_queue_rules_and_route(): routing_client = TestClient(routing_app) created = routing_client.post( "/queues", json={ "name": "General Queue", "description": "Main line", "rules": [ { "channel": "voice", "priority": 2, "strategy": "round_robin", "sla_seconds": 45, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert created.status_code == 200 queue_id = created.json()["queue_id"] route = routing_client.post( f"/queues/{queue_id}/route?channel=voice&priority=2", headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert route.status_code == 200 assert route.json()["sla_seconds"] == 45 def test_list_queues_tolerates_legacy_rules_json_object(): routing_client = TestClient(routing_app) session = get_session() try: session.add( Queue( queue_id="que_legacy_rules", name="Legacy Queue", description="Created before rules_json normalization", rules_json="{}", created_at="2026-03-15T00:00:00+00:00", ) ) session.commit() finally: session.close() listed = routing_client.get( "/queues", headers={"X-User": "admin", "X-Role": "admin"}, ) assert listed.status_code == 200 payload = listed.json() queue = next(item for item in payload if item["queue_id"] == "que_legacy_rules") assert queue["rules"] == [] def test_queue_route_requires_privileged_role(): routing_client = TestClient(routing_app) created = routing_client.post( "/queues", json={ "name": "Guarded Queue", "description": "RBAC", "rules": [ { "channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert created.status_code == 200 denied = routing_client.post( f"/queues/{created.json()['queue_id']}/route?channel=voice&priority=3", headers={"X-User": "operator", "X-Role": "operator"}, ) assert denied.status_code == 403 def test_admin_can_delete_queue(): routing_client = TestClient(routing_app) created = routing_client.post( "/queues", json={ "name": "Disposable Queue", "description": "To delete", "rules": [ { "channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert created.status_code == 200 queue_id = created.json()["queue_id"] deleted = routing_client.delete( f"/queues/{queue_id}", headers={"X-User": "admin", "X-Role": "admin"}, ) assert deleted.status_code == 200 assert deleted.json()["deleted"] is True route = routing_client.post( f"/queues/{queue_id}/route?channel=voice&priority=3", headers={"X-User": "admin", "X-Role": "admin"}, ) assert route.status_code == 404 def test_supervisor_cannot_delete_queue(): routing_client = TestClient(routing_app) created = routing_client.post( "/queues", json={ "name": "Protected Queue", "description": "RBAC delete", "rules": [ { "channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert created.status_code == 200 denied = routing_client.delete( f"/queues/{created.json()['queue_id']}", headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert denied.status_code == 403 def test_route_supports_completed_ivr_session_and_rejects_incomplete_session(): routing_client = TestClient(routing_app) ivr_client = TestClient(ivr_app) origin_queue = routing_client.post( "/queues", json={ "name": "IVR Origin Queue", "description": "voice origin", "rules": [ { "channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert origin_queue.status_code == 200 origin_queue_id = origin_queue.json()["queue_id"] resolved_queue = routing_client.post( "/queues", json={ "name": "IVR Resolved Queue", "description": "voice support", "rules": [ { "channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 55, } ], }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert resolved_queue.status_code == 200 resolved_queue_id = resolved_queue.json()["queue_id"] flow = ivr_client.post( "/ivr/flows", json={ "name": "Route IVR", "description": "route override", "queue_id": origin_queue_id, "entry_node_id": "root", "flow_json": { "nodes": [ { "node_id": "root", "prompt_text": "Press 2 for support", "is_terminal": False, "invalid_target_node_id": "root", "options": [{"digit": "2", "target_node_id": "support"}], }, { "node_id": "support", "prompt_text": "Support queue", "is_terminal": True, "outcome_code": "support_route", "resolved_queue_id": resolved_queue_id, "resolved_queue_code": "support", "options": [], }, ] }, "is_active": True, }, headers={"X-User": "admin", "X-Role": "admin"}, ) assert flow.status_code == 200 active_session = ivr_client.post( "/ivr/sessions/start", json={"call_id": "call_route_ivr", "queue_id": origin_queue_id}, headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert active_session.status_code == 200 session_id = active_session.json()["session"]["session_id"] incomplete = routing_client.post( f"/queues/{origin_queue_id}/route?channel=voice&priority=3&ivr_session_id={session_id}", headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert incomplete.status_code == 400 stepped = ivr_client.post( f"/ivr/sessions/{session_id}/dtmf", json={"digit": "2"}, headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert stepped.status_code == 200 assert stepped.json()["session"]["status"] == "completed" routed = routing_client.post( f"/queues/{origin_queue_id}/route?channel=voice&priority=3&ivr_session_id={session_id}", headers={"X-User": "supervisor", "X-Role": "supervisor"}, ) assert routed.status_code == 200 payload = routed.json() assert payload["queue_id"] == resolved_queue_id assert payload["original_queue_id"] == origin_queue_id assert payload["resolved_queue_id"] == resolved_queue_id assert payload["ivr_session_id"] == session_id assert payload["ivr_outcome_code"] == "support_route" assert payload["sla_seconds"] == 55