core-python/tests/test_voice_realtime.py

"""Tests realtime voice assistant arhitektūrai."""

from __future__ import annotations

import pytest

from maris_core.voice.livekit_agent import (
    build_livekit_agent_metadata,
    load_livekit_voice_agent_settings,
    require_livekit_agents_runtime,
)
from maris_core.voice.realtime import RealtimeVoiceAssistant, VoiceSessionState


class _FakeSttProvider:
    def stream_transcript(self, audio_chunks, *, language: str, session_id: str):
        del audio_chunks, language, session_id
        yield {"text": "Sveiki", "is_final": False}
        yield {"text": "Sveiki, palīdzi man", "is_final": True}


class _FakeLlmProvider:
    def stream_response(self, transcript: str, *, session_id: str, persona_id: str | None):
        del transcript, session_id, persona_id
        yield "Labdien! "
        yield "Es palīdzēšu."


class _FakeTtsProvider:
    def stream_audio(self, text_chunks, *, language: str, voice: str, session_id: str):
        del language, voice, session_id
        for index, chunk in enumerate(text_chunks):
            yield {"audio": chunk.encode("utf-8"), "mime_type": "audio/wav", "is_final": index == 1}


def test_load_livekit_voice_agent_settings_and_metadata() -> None:
    settings = load_livekit_voice_agent_settings(
        {
            "LIVEKIT_URL": "wss://livekit.example",
            "LIVEKIT_API_KEY": "key",
            "LIVEKIT_API_SECRET": "secret",
            "MARIS_VOICE_STT_PROVIDER": "speechmatics",
            "MARIS_VOICE_TTS_PROVIDER": "azure",
        }
    )
    metadata = build_livekit_agent_metadata(settings)

    assert settings.livekit_url == "wss://livekit.example"
    assert metadata["runtime"] == "livekit-agents"
    assert metadata["features"]["barge_in"] is True


def test_realtime_voice_assistant_streams_full_turn() -> None:
    assistant = RealtimeVoiceAssistant(
        stt=_FakeSttProvider(),
        llm=_FakeLlmProvider(),
        tts=_FakeTtsProvider(),
    )

    events = list(
        assistant.handle_turn([b"audio"], session=VoiceSessionState(session_id="voice-1"))
    )

    assert events[0]["type"] == "vad"
    assert any(event["type"] == "transcript" and event["is_final"] for event in events)
    assert any(event["type"] == "llm_delta" for event in events)
    assert any(event["type"] == "tts_audio" for event in events)
    assert events[-1]["type"] == "complete"
    assert events[-1]["text"] == "Labdien! Es palīdzēšu."


def test_realtime_voice_assistant_supports_barge_in() -> None:
    assistant = RealtimeVoiceAssistant(
        stt=_FakeSttProvider(),
        llm=_FakeLlmProvider(),
        tts=_FakeTtsProvider(),
    )
    session = VoiceSessionState(session_id="voice-2")
    iterator = assistant.handle_turn([b"audio"], session=session)

    next(iterator)  # vad
    next(iterator)  # partial transcript
    next(iterator)  # final transcript
    first_llm_delta = next(iterator)
    interrupt_event = assistant.interrupt(session)
    barge_in_event = next(iterator)

    assert first_llm_delta["type"] == "llm_delta"
    assert interrupt_event["type"] == "barge_in"
    assert barge_in_event["type"] == "barge_in"
    assert barge_in_event["interrupted"] is True


def test_require_livekit_agents_runtime_raises_clear_error() -> None:
    with pytest.raises(RuntimeError) as exc_info:
        require_livekit_agents_runtime()

    assert "LiveKit Agents runtime nav instalēts" in str(exc_info.value)