""" Unit tests for app.internal.ai_health — the shared AI-provider degradation registry added for logan/server-26#14 (no alerting when a provider account runs dry or a model is retired). Covers: * classify() telling a permanent condition (dead model, depleted billing — both of which can arrive as the same HTTP status a rate limit uses) apart from a transient one. * report_degraded() alerting immediately for a permanent condition but only after TRANSIENT_ALERT_THRESHOLD consecutive failures for a transient one. * Alerting exactly once per episode, not once per call, and again exactly once on recovery. * report_healthy() clearing degraded state so a later re-degradation can alert again (a fresh episode, not a continuation of the old one). The Discord webhook is patched at ai_health._post_webhook so no real HTTP is made; settings.ai_alert_webhook_url is irrelevant to these tests since _post_webhook itself is replaced. """ import pytest from unittest.mock import AsyncMock, patch from app.internal import ai_health @pytest.fixture(autouse=True) def _reset_state(): """Every test gets a clean registry — module-level state persists otherwise.""" ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} yield ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} # --------------------------------------------------------------------------- # classify() # --------------------------------------------------------------------------- def test_classify_dead_model_404(): assert ai_health.classify("404 models/gemini-2.0-flash is not found") == "dead_model" def test_classify_dead_model_no_longer_available(): assert ai_health.classify("this model is no longer available") == "dead_model" def test_classify_billing_depleted_credits(): # The exact wording that bit the Gemini correlator on 2026-08-18. assert ai_health.classify("429 prepayment credits are depleted") == "billing" def test_classify_billing_openai_insufficient_quota(): assert ai_health.classify("Error: insufficient_quota — exceeded your current quota") == "billing" def test_classify_ordinary_rate_limit_is_transient(): # Same HTTP status (429) as the depleted-balance case, but no billing # wording — this must NOT be classified as billing. assert ai_health.classify("429 Too Many Requests, please retry later") == "transient" def test_classify_network_error_is_transient(): assert ai_health.classify("Connection reset by peer") == "transient" # --------------------------------------------------------------------------- # report_degraded — permanent alerts immediately # --------------------------------------------------------------------------- async def test_permanent_failure_alerts_on_first_occurrence(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: await ai_health.report_degraded( "correlation_cheap", "gemini", "gemini-2.0-flash", "model is unavailable", "update the model ID", permanent=True, ) webhook.assert_awaited_once() state = ai_health.snapshot()["correlation_cheap"] assert state["degraded"] is True assert state["alerted"] is True assert state["permanent"] is True async def test_permanent_failure_alerts_only_once_per_episode(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: for _ in range(5): await ai_health.report_degraded( "correlation_cheap", "gemini", "gemini-2.0-flash", "model is unavailable", "update the model ID", permanent=True, ) # Once per episode, not once per call — this runs at radio-traffic volume. webhook.assert_awaited_once() assert ai_health.snapshot()["correlation_cheap"]["consecutive_failures"] == 5 # --------------------------------------------------------------------------- # report_degraded — transient only alerts once it persists # --------------------------------------------------------------------------- async def test_transient_failure_does_not_alert_below_threshold(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: for _ in range(ai_health.TRANSIENT_ALERT_THRESHOLD - 1): await ai_health.report_degraded( "transcription", "openai", "whisper-1", "transient API error", "no action needed unless this persists", permanent=False, ) webhook.assert_not_awaited() state = ai_health.snapshot()["transcription"] assert state["degraded"] is False assert state["alerted"] is False async def test_transient_failure_alerts_once_threshold_crossed(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: for _ in range(ai_health.TRANSIENT_ALERT_THRESHOLD): await ai_health.report_degraded( "transcription", "openai", "whisper-1", "transient API error", "no action needed unless this persists", permanent=False, ) webhook.assert_awaited_once() # Further failures in the same episode must not re-alert. await ai_health.report_degraded( "transcription", "openai", "whisper-1", "transient API error", "no action needed unless this persists", permanent=False, ) webhook.assert_awaited_once() # --------------------------------------------------------------------------- # report_healthy — recovery # --------------------------------------------------------------------------- async def test_recovery_alerts_once_after_an_alerted_episode(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: await ai_health.report_degraded( "correlation_smart", "gemini", "gemini-1.5-pro", "the Gemini account is out of credit", "top up billing", permanent=True, ) webhook.reset_mock() await ai_health.report_healthy("correlation_smart") webhook.assert_awaited_once() state = ai_health.snapshot()["correlation_smart"] assert state["degraded"] is False assert state["alerted"] is False assert state["consecutive_failures"] == 0 async def test_recovery_from_never_alerted_transient_state_is_silent(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: # Two failures — below the transient threshold, never alerted. await ai_health.report_degraded( "extraction", "gemini", "gemini-3.6-flash", "transient API error", "no action needed unless this persists", permanent=False, ) await ai_health.report_degraded( "extraction", "gemini", "gemini-3.6-flash", "transient API error", "no action needed unless this persists", permanent=False, ) webhook.reset_mock() await ai_health.report_healthy("extraction") # Nothing was ever posted for this episode, so recovery posts nothing either. webhook.assert_not_awaited() async def test_recovery_then_re_degradation_alerts_again_as_a_new_episode(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: await ai_health.report_degraded( "correlation_cheap", "gemini", "gemini-2.0-flash", "model is unavailable", "update the model ID", permanent=True, ) await ai_health.report_healthy("correlation_cheap") webhook.reset_mock() # A second, later episode must alert on its own first occurrence. await ai_health.report_degraded( "correlation_cheap", "gemini", "gemini-2.0-flash", "model is unavailable", "update the model ID", permanent=True, ) webhook.assert_awaited_once() # --------------------------------------------------------------------------- # snapshot() # --------------------------------------------------------------------------- async def test_snapshot_reports_all_tiers_healthy_by_default(): state = ai_health.snapshot() assert set(state.keys()) == set(ai_health.TIERS) for tier_state in state.values(): assert tier_state["degraded"] is False assert tier_state["consecutive_failures"] == 0