my_voice_assistant_v3/tests/test_emergency_llm.py

import asyncio

import pytest

from app.config import settings
from app.safety import emergency as em
from app.safety import llm_classifier as lc


class StubLLM:
    def __init__(self, raw):
        self.raw = raw
        self.calls = 0

    async def complete(self, text, history=None, session_id=None, language=None):
        self.calls += 1
        return self.raw


class FakeUser:
    id = "anonymous"
    display_name = "Test"


class FakeStore:
    def __init__(self):
        self.logged = []

    def log_emergency(self, user_id, category, snippet):
        self.logged.append((user_id, category, snippet))


@pytest.fixture(autouse=True)
def _clear_pending():
    em._pending.clear()
    yield
    em._pending.clear()


def test_parse_classification_variants():
    assert lc.parse_classification('{"category":"medical","confidence":0.9}')["category"] == "medical"
    # umschlossen von Text
    got = lc.parse_classification('Antwort: {"category":"help","confidence":0.7,"reason":"Feuer"}')
    assert got["category"] == "help" and got["confidence"] == 0.7
    # 'none' ist kein Notfall
    assert lc.parse_classification('{"category":"none","confidence":0.99}') is None
    # unbekannte Kategorie
    assert lc.parse_classification('{"category":"foo","confidence":0.99}') is None
    assert lc.parse_classification("kein json") is None
    assert lc.parse_classification("") is None


def test_classify_respects_confidence_threshold(monkeypatch):
    monkeypatch.setattr(settings, "emergency_llm_min_confidence", 0.6)
    monkeypatch.setattr(lc, "_build_classifier_llm",
                        lambda cfg: StubLLM('{"category":"medical","confidence":0.4}'))
    # unter der Schwelle -> kein Notfall
    assert asyncio.run(lc.classify_emergency("mir ist schwindelig")) is None

    monkeypatch.setattr(lc, "_build_classifier_llm",
                        lambda cfg: StubLLM('{"category":"medical","confidence":0.85}'))
    res = asyncio.run(lc.classify_emergency("mir wird ganz schwarz vor augen"))
    assert res["category"] == "medical"


def test_schedule_skips_when_keyword_already_hit(monkeypatch):
    store = FakeStore()
    called = StubLLM('{"category":"medical","confidence":0.9}')
    monkeypatch.setattr(lc, "_build_classifier_llm", lambda cfg: called)

    async def run():
        # keyword_hit ist gesetzt -> Stufe 2 wird uebersprungen
        task = em.schedule_llm_emergency_check(
            FakeUser(), "egal", store, {"category": "medical", "matched": "x"}
        )
        return task

    assert asyncio.run(run()) is None
    assert called.calls == 0
    assert store.logged == []


def test_schedule_disabled_is_noop(monkeypatch):
    monkeypatch.setattr(settings, "emergency_llm_enabled", False)
    store = FakeStore()

    async def run():
        return em.schedule_llm_emergency_check(FakeUser(), "hilfe", store, None)

    assert asyncio.run(run()) is None
    assert store.logged == []


def test_schedule_escalates_and_calls_callback(monkeypatch):
    monkeypatch.setattr(settings, "emergency_llm_enabled", True)
    monkeypatch.setattr(settings, "emergency_llm_min_confidence", 0.6)
    monkeypatch.setattr(lc, "_build_classifier_llm",
                        lambda cfg: StubLLM('{"category":"self_harm","confidence":0.95}'))
    store = FakeStore()
    events = []

    async def on_emergency(category):
        events.append(category)

    async def run():
        task = em.schedule_llm_emergency_check(
            FakeUser(), "ich will nicht mehr weiterleben",
            store, None, on_emergency=on_emergency,
        )
        await task

    asyncio.run(run())
    assert store.logged == [("anonymous", "self_harm", "ich will nicht mehr weiterleben")]
    assert events == ["self_harm"]


def test_schedule_malformed_output_no_escalation(monkeypatch):
    monkeypatch.setattr(settings, "emergency_llm_enabled", True)
    monkeypatch.setattr(lc, "_build_classifier_llm",
                        lambda cfg: StubLLM("Tut mir leid, kein JSON."))
    store = FakeStore()

    async def run():
        task = em.schedule_llm_emergency_check(FakeUser(), "irgendwas", store, None)
        await task

    asyncio.run(run())
    assert store.logged == []