cf-voice/tests/test_telephony.py

import asyncio
import pytest
from cf_voice.telephony import (
    CallSession,
    MockTelephonyBackend,
    TelephonyBackend,
    make_telephony,
)


class TestCallSession:
    def test_defaults(self):
        s = CallSession(call_sid="sid_1", to="+15551234567", from_="+18005550000")
        assert s.state == "dialing"
        assert s.amd_result == "unknown"
        assert s.duration_s == 0.0
        assert s.error is None

    def test_state_mutation(self):
        s = CallSession(call_sid="sid_2", to="+1", from_="+2", state="in_progress")
        s.state = "completed"
        assert s.state == "completed"


class TestMockTelephonyBackend:
    @pytest.mark.asyncio
    async def test_dial_returns_session(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+15551234567", "+18005550000", "https://example.com/wh")
        assert isinstance(session, CallSession)
        assert session.call_sid.startswith("mock_sid_")
        assert session.to == "+15551234567"
        assert session.from_ == "+18005550000"

    @pytest.mark.asyncio
    async def test_dial_transitions_to_in_progress(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+15551234567", "+18005550000", "https://x.com")
        # give the background task a moment to transition
        await asyncio.sleep(0.1)
        assert session.state == "in_progress"

    @pytest.mark.asyncio
    async def test_amd_resolves_human(self):
        backend = MockTelephonyBackend(amd_delay_s=0.05)
        session = await backend.dial("+1555", "+1800", "https://x.com", amd=True)
        await asyncio.sleep(0.2)
        assert session.amd_result == "human"

    @pytest.mark.asyncio
    async def test_send_dtmf(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        # should not raise
        await backend.send_dtmf(session.call_sid, "1234#")

    @pytest.mark.asyncio
    async def test_send_dtmf_unknown_sid_raises(self):
        backend = MockTelephonyBackend()
        with pytest.raises(KeyError):
            await backend.send_dtmf("nonexistent_sid", "1")

    @pytest.mark.asyncio
    async def test_bridge_updates_state(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        await backend.bridge(session.call_sid, "+15559999999")
        assert session.state == "bridged"

    @pytest.mark.asyncio
    async def test_hangup_sets_completed(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        await backend.hangup(session.call_sid)
        assert session.state == "completed"

    @pytest.mark.asyncio
    async def test_hangup_idempotent(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        await backend.hangup(session.call_sid)
        await backend.hangup(session.call_sid)
        assert session.state == "completed"

    @pytest.mark.asyncio
    async def test_announce_does_not_raise(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        await backend.announce(session.call_sid, "Hello, this is an automated assistant.")

    @pytest.mark.asyncio
    async def test_get_state(self):
        backend = MockTelephonyBackend()
        session = await backend.dial("+1", "+2", "https://x.com")
        state = await backend.get_state(session.call_sid)
        assert state in ("ringing", "in_progress", "dialing")

    @pytest.mark.asyncio
    async def test_multiple_calls_unique_sids(self):
        backend = MockTelephonyBackend()
        s1 = await backend.dial("+1", "+2", "https://x.com")
        s2 = await backend.dial("+3", "+4", "https://x.com")
        assert s1.call_sid != s2.call_sid

    def test_isinstance_protocol(self):
        backend = MockTelephonyBackend()
        assert isinstance(backend, TelephonyBackend)


class TestMakeTelephony:
    def test_mock_flag(self):
        backend = make_telephony(mock=True)
        assert isinstance(backend, MockTelephonyBackend)

    def test_mock_env(self, monkeypatch):
        monkeypatch.setenv("CF_VOICE_MOCK", "1")
        backend = make_telephony()
        assert isinstance(backend, MockTelephonyBackend)

    def test_no_config_raises(self, monkeypatch):
        monkeypatch.delenv("CF_VOICE_MOCK", raising=False)
        monkeypatch.delenv("CF_SW_PROJECT_ID", raising=False)
        monkeypatch.delenv("CF_ESL_PASSWORD", raising=False)
        with pytest.raises(RuntimeError, match="No telephony backend configured"):
            make_telephony()

    def test_signalwire_selected_by_env(self, monkeypatch):
        monkeypatch.delenv("CF_VOICE_MOCK", raising=False)
        monkeypatch.setenv("CF_SW_PROJECT_ID", "proj_123")
        # SignalWireBackend will raise ImportError (signalwire SDK not installed)
        # but only at instantiation — make_telephony should call the constructor
        with pytest.raises((ImportError, RuntimeError)):
            make_telephony()

    def test_freeswitch_selected_by_env(self, monkeypatch):
        monkeypatch.delenv("CF_VOICE_MOCK", raising=False)
        monkeypatch.delenv("CF_SW_PROJECT_ID", raising=False)
        monkeypatch.setenv("CF_ESL_PASSWORD", "s3cret")
        # FreeSWITCHBackend will raise ImportError (ESL not installed)
        with pytest.raises((ImportError, RuntimeError)):
            make_telephony()