hold-slayer/tests/test_graceful_degradation.py

"""
Graceful-degradation tests.

Every external dependency — STT, LLM, TTS — is reachable over the network and
can be down. The gateway's rule is that a dead dependency degrades the call
rather than killing it, and says so: each failure publishes an `ERROR` event
naming the service, so a down Speaches reads as "transcription failed" rather
than "the AI is making bad decisions".

The behaviour is already implemented across the services; these tests exist so a
later refactor can't quietly remove it. The failure mode being guarded against is
silent: an un-caught exception in one of these paths aborts a live phone call.
"""

from unittest.mock import AsyncMock, MagicMock

import pytest

from config import Settings
from core.event_bus import EventBus
from models.events import EventType


def _gateway():
    """A gateway stand-in with a real event bus, so events can be asserted."""
    gw = MagicMock()
    gw.settings = Settings(database_url="sqlite+aiosqlite:///:memory:")
    gw.event_bus = EventBus()
    gw.call_manager = MagicMock()
    gw.call_manager.add_transcript = AsyncMock()
    return gw


def _hold_slayer(gateway, transcription):
    from services.audio_classifier import AudioClassifier
    from services.hold_slayer import HoldSlayerService

    return HoldSlayerService(
        gateway=gateway,
        call_manager=gateway.call_manager,
        sip_engine=MagicMock(),
        classifier=AudioClassifier(gateway.settings.classifier),
        transcription=transcription,
        settings=gateway.settings,
    )


async def _errors_for(bus: EventBus, coro):
    """Run `coro` while subscribed, returning the ERROR events it published."""
    import asyncio

    sub = bus.subscribe(event_types={EventType.ERROR})
    try:
        result = await coro
        seen = []
        while True:
            try:
                seen.append(sub._queue.get_nowait())
            except asyncio.QueueEmpty:
                break
        return result, seen
    finally:
        bus.unsubscribe(sub)


class TestClassifierWithoutSTT:
    """The classifier is spectral: it must not depend on STT at all."""

    def _classifier(self):
        from services.audio_classifier import AudioClassifier

        return AudioClassifier(Settings(database_url="sqlite+aiosqlite:///:memory:").classifier)

    def test_classify_takes_only_audio(self):
        # A transcript parameter would make STT a hard dependency of hold
        # detection — the thing this checkbox is about.
        import inspect

        params = set(inspect.signature(self._classifier().classify_chunk).parameters)
        assert params == {"audio_data"}

    async def test_classifies_with_no_stt_service_anywhere(self):
        # Silence is the cheapest deterministic input; the point is that a
        # classification is produced at all with no STT in the picture.
        result = await self._classifier().classify(b"\x00\x00" * 16000)
        assert result.audio_type is not None


class TestTranscriptionDegradation:
    async def test_hold_slayer_transcribe_returns_empty_on_failure(self):
        gw = _gateway()
        stt = MagicMock()
        stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
        svc = _hold_slayer(gw, stt)

        text, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))

        assert text == ""  # empty transcript, not an exception
        assert len(errors) == 1
        assert errors[0].data["service"] == "transcription"

    async def test_error_event_names_the_service(self):
        # "transcription failed" vs "the AI decided badly" — the whole reason
        # transcribe() raises instead of swallowing.
        gw = _gateway()
        stt = MagicMock()
        stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
        svc = _hold_slayer(gw, stt)

        _, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
        assert "Connection refused" in errors[0].data["error"]

    async def test_service_error_survives_a_dead_event_bus(self):
        # Degradation reporting must not itself become a failure path.
        gw = _gateway()
        gw.event_bus.publish = AsyncMock(side_effect=RuntimeError("bus down"))
        stt = MagicMock()
        stt.transcribe = AsyncMock(side_effect=RuntimeError("stt down"))
        svc = _hold_slayer(gw, stt)

        assert await svc._transcribe("call-1", b"\x00" * 320) == ""

    async def test_transcription_marks_itself_unavailable(self):
        # /health reads this flag; a failure that doesn't record itself makes
        # the probe lie.
        import httpx

        from services.transcription import TranscriptionService

        svc = TranscriptionService(Settings(database_url="sqlite+aiosqlite:///:memory:").speaches)
        client = MagicMock()
        client.post = AsyncMock(side_effect=httpx.ConnectError("refused"))
        svc._client = client
        svc._client.is_closed = False

        with pytest.raises(Exception):
            await svc.transcribe(b"\x00" * 320)
        assert svc.available is False


class TestReceptionistDegradation:
    def _receptionist(self, gateway, **kw):
        from services.receptionist import ReceptionistService

        return ReceptionistService(gateway=gateway, **kw)

    async def test_llm_failure_falls_back_to_a_usable_decision(self):
        gw = _gateway()
        svc = self._receptionist(gw)
        call = MagicMock(id="call-1", remote_number="+15551234567")

        llm = MagicMock()
        llm.chat_json = AsyncMock(side_effect=RuntimeError("LLM down"))
        import services.llm_client as llm_mod

        original = llm_mod.get_llm
        llm_mod.get_llm = lambda: llm
        try:
            result, errors = await _errors_for(
                gw.event_bus, svc._classify(call, "I need to speak to someone", None)
            )
        finally:
            llm_mod.get_llm = original

        # A decision still comes back, so the call can proceed.
        assert result["recommended_action"] in {"ring", "message", "reject"}
        assert errors[0].data["service"] == "llm"

    async def test_no_transcription_service_yields_empty_not_crash(self):
        # transcription=None is a valid wiring (STT not configured).
        gw = _gateway()
        svc = self._receptionist(gw, transcription=None)
        assert svc.transcription is None


class TestHealthReportsDegradation:
    """A degraded gateway must read as degraded — /health may not lie."""

    def test_availability_helper_distinguishes_unknown_from_down(self):
        # Four distinct states, because "not wired up" and "wired up but the
        # remote is refusing connections" are different operator problems.
        import main

        assert main._availability(None) == "not attached"
        assert main._availability(MagicMock(available=None)) == "unknown (no requests yet)"
        assert main._availability(MagicMock(available=True)) == "ok"
        assert main._availability(MagicMock(available=False)) == "unreachable"