call-center/tests/test_escalation_retry.py

import json

import pytest

import services.asterisk_bridge_service.voice_ai as voice_ai
from services.routing_service import engine as routing_engine
from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import AgentRow, AsteriskCallLinkRow, EscalationRow


def _make_agent(session, *, extension: str, status: str = "AVAILABLE"):
    now = utc_now_iso()
    row = AgentRow(
        agent_id=new_id("agt"),
        tenant_ids_json="[]",
        extension=extension,
        endpoint=None,
        display_name=f"Agent {extension}",
        level="L2",
        skills_json="[]",
        status=status,
        max_concurrent_calls=1,
        enabled=True,
        created_at=now,
        updated_at=now,
    )
    session.add(row)
    session.commit()
    session.refresh(row)
    return row


def _make_call_link(session, *, call_id: str):
    now = utc_now_iso()
    link = AsteriskCallLinkRow(
        call_id=call_id,
        linked_id=call_id,
        queue_code="voice_lab_ai",
        queue_id="que_test",
        interaction_id="int_test",
        status="active",
        telephony_status="ringing",
        channel_name=f"PJSIP/tele2-kazgaz-{call_id}",
        current_level="L2",
        started_at=now,
        updated_at=now,
    )
    session.add(link)
    session.commit()
    session.refresh(link)
    return link


def _make_ringing_escalation(session, *, call_id: str, agent: AgentRow):
    now = utc_now_iso()
    escalation = EscalationRow(
        escalation_id=new_id("esc"),
        call_id=call_id,
        tenant_id=None,
        from_level="L1",
        to_level="L2",
        reason_code="AI_UNABLE_TO_RESOLVE",
        required_skills_json="[]",
        priority=3,
        status="ringing",
        assigned_agent_id=agent.extension,
        real_agent_id=agent.agent_id,
        attempted_agent_ids_json=json.dumps([agent.agent_id]),
        requested_at=now,
    )
    session.add(escalation)
    session.commit()
    session.refresh(escalation)
    return escalation


def _patch_routing_over_http(monkeypatch, session):
    """Make voice_ai's HTTP-facing routing helpers operate on the same test session directly."""

    def fake_reserve(*, call_id, level, tenant_id, required_skills=None, exclude_agent_ids=None):
        agent = routing_engine.reserve_agent(
            session,
            call_id=call_id,
            level=level,
            tenant_id=tenant_id,
            required_skills=required_skills,
            exclude_agent_ids=exclude_agent_ids,
        )
        if agent is None:
            return None
        return {
            "agent_id": agent.agent_id,
            "extension": agent.extension,
            "endpoint": agent.endpoint,
            "display_name": agent.display_name,
        }

    def fake_release_by_agent_id(agent_id, *, next_status="AVAILABLE"):
        routing_engine.release_agent_by_id(session, agent_id=agent_id, next_status=next_status)

    def fake_set_status(agent_id, status):
        routing_engine.set_agent_status(session, agent_id=agent_id, status=status)

    monkeypatch.setattr(voice_ai, "_reserve_routing_agent", fake_reserve)
    monkeypatch.setattr(voice_ai, "routing_release_by_agent_id", fake_release_by_agent_id)
    monkeypatch.setattr(voice_ai, "set_routing_agent_status", fake_set_status)
    monkeypatch.setattr(voice_ai, "_append_escalation_timeline", lambda *a, **k: None)
    monkeypatch.setattr(voice_ai, "_emit_escalation_event", lambda *a, **k: None)
    monkeypatch.setattr(voice_ai, "_resolve_handoff_channel", lambda session, link: link.channel_name)


def test_retry_escalation_no_answer_moves_to_next_available_agent(monkeypatch):
    init_sql_schema()
    redirected_to: list[str] = []

    session = get_session()
    try:
        _patch_routing_over_http(monkeypatch, session)
        monkeypatch.setattr(
            voice_ai,
            "_redirect_channel_to_agent",
            lambda *, channel, extension: redirected_to.append(extension),
        )
        agent_a = _make_agent(session, extension="2001")
        agent_b = _make_agent(session, extension="2002")
        routing_engine.reserve_agent(session, call_id="call-retry-1", level="L2", tenant_id=None, required_skills=[])
        link = _make_call_link(session, call_id="call-retry-1")
        escalation = _make_ringing_escalation(session, call_id="call-retry-1", agent=agent_a)

        voice_ai.retry_escalation_no_answer(session, call_id="call-retry-1", dial_outcome="NOANSWER")

        session.refresh(escalation)
        session.refresh(agent_a)
        session.refresh(agent_b)

        assert escalation.attempt_count == 1
        assert escalation.status == "ringing"
        assert escalation.real_agent_id == agent_b.agent_id
        assert escalation.assigned_agent_id == agent_b.extension
        assert json.loads(escalation.attempted_agent_ids_json) == [agent_a.agent_id, agent_b.agent_id]

        assert agent_a.status == "AVAILABLE"
        assert agent_a.current_call_id is None
        assert agent_b.status == "RINGING"
        assert agent_b.current_call_id == "call-retry-1"

        assert redirected_to == [agent_b.extension]
    finally:
        session.close()


def test_retry_escalation_no_answer_exhausts_pool_marks_failed(monkeypatch):
    init_sql_schema()

    session = get_session()
    try:
        _patch_routing_over_http(monkeypatch, session)
        monkeypatch.setattr(voice_ai, "_redirect_channel_to_agent", lambda **kwargs: None)
        agent_a = _make_agent(session, extension="3001")
        routing_engine.reserve_agent(session, call_id="call-retry-2", level="L2", tenant_id=None, required_skills=[])
        link = _make_call_link(session, call_id="call-retry-2")
        escalation = _make_ringing_escalation(session, call_id="call-retry-2", agent=agent_a)

        voice_ai.retry_escalation_no_answer(session, call_id="call-retry-2", dial_outcome="NOANSWER")

        session.refresh(escalation)
        session.refresh(agent_a)

        assert escalation.attempt_count == 1
        assert escalation.status == "failed"
        assert agent_a.status == "AVAILABLE"
    finally:
        session.close()


def test_retry_escalation_no_answer_ignores_calls_without_ringing_escalation(monkeypatch):
    init_sql_schema()
    called = []

    session = get_session()
    try:
        _patch_routing_over_http(monkeypatch, session)
        monkeypatch.setattr(voice_ai, "_redirect_channel_to_agent", lambda **kwargs: called.append(kwargs))
        _make_call_link(session, call_id="call-no-escalation")
        voice_ai.retry_escalation_no_answer(session, call_id="call-no-escalation", dial_outcome="NOANSWER")
        assert called == []
    finally:
        session.close()