voiceagent/tests/test_protocol.py

"""
Tests for the Voice Assistant WebSocket protocol.

Purpose:
    Verify that protocol message models enforce their declared
    event types and canonical audio contract.

Responsibilities:
    - Validate inbound protocol event discriminators.
    - Validate SessionStart audio configuration.
    - Reject messages whose type does not match their model.

Non-responsibilities:
    - Testing WebSocket transport behavior.
    - Testing session or turn lifecycle.
    - Testing STT, Agent, or TTS providers.
"""

import pytest
from pydantic import ValidationError
from voice_assistant.audio.models import CANONICAL_AUDIO_FORMAT
from voice_assistant.audio_gateway.app.protocol import (
    AgentDelta,
    AgentResponse,
    AudioChunk,
    AudioStart,
    AudioStop,
    ErrorEvent,
    Interrupt,
    Interrupted,
    SessionEnd,
    SessionEnded,
    SessionReady,
    SessionStart,
    STTFinal,
    STTPartial,
    TurnEnded,
    TurnStarted,
)



@pytest.mark.parametrize(
    ("model", "expected_type"),
    [
        (SessionStart, "session.start"),
        (AudioStart, "audio.start"),
        (AudioChunk, "audio.chunk"),
        (AudioStop, "audio.stop"),
        (Interrupt, "interrupt"),
        (SessionEnd, "session.end"),
    ],
)
def test_protocol_event_has_expected_type(
    model,
    expected_type,
):
    payload = {
        "type": expected_type,
    }

    if model in (AudioChunk, AudioStop, Interrupt):
        payload["turn_id"] = 1

    if model is AudioChunk:
        payload["sequence_number"] = 0

    event = model.model_validate(payload)

    assert event.type == expected_type


def test_protocol_rejects_wrong_event_type():
    with pytest.raises(ValidationError):
        AudioStart.model_validate(
            {
                "type": "audio.stop",
            }
        )

def test_session_ready_output():
    event = SessionReady(
        protocol_version="1.0",
        session_id="session-1",
        audio=CANONICAL_AUDIO_FORMAT,
    )

    assert event.model_dump() == {
        "type": "session.ready",
        "protocol_version": "1.0",
        "session_id": "session-1",
        "audio": {
            "encoding": "pcm_s16le",
            "sample_rate": 16000,
            "channels": 1,
        },
    }


@pytest.mark.parametrize(
    ("model", "event_type"),
    [
        (TurnStarted, "turn.started"),
        (STTPartial, "stt.partial"),
        (STTFinal, "stt.final"),
        (AgentDelta, "agent.delta"),
        (AgentResponse, "agent.response"),
        (TurnEnded, "turn.ended"),
        (Interrupted, "interrupted"),
    ],
)
def test_turn_output_event_types(
    model,
    event_type,
):
    data = {
        "session_id": "session-1",
        "turn_id": 1,
    }

    if model in (STTPartial, STTFinal, AgentDelta, AgentResponse):
        data["text"] = "test"

    event = model(**data)

    assert event.type == event_type


def test_session_ended_output():
    event = SessionEnded(
        session_id="session-1",
    )

    assert event.type == "session.ended"
    assert event.session_id == "session-1"

def test_error_event():
    event = ErrorEvent(
        session_id="session-1",
        turn_id=1,
        code="TEST_ERROR",
        message="Something went wrong.",
        retryable=True,
        details={
            "reason": "test",
        },
    )

    assert event.model_dump() == {
        "type": "error",
        "session_id": "session-1",
        "turn_id": 1,
        "code": "TEST_ERROR",
        "message": "Something went wrong.",
        "retryable": True,
        "details": {
            "reason": "test",
        },
    }


def test_error_event_defaults():
    event = ErrorEvent(
        code="TEST_ERROR",
        message="Something went wrong.",
    )

    assert event.type == "error"
    assert event.session_id is None
    assert event.turn_id is None
    assert event.retryable is False
    assert event.details == {}


def test_agent_delta_serialization():
    event = AgentDelta(
        session_id="session-1",
        turn_id=7,
        text=" mundo",
    )

    assert event.model_dump() == {
        "type": "agent.delta",
        "session_id": "session-1",
        "turn_id": 7,
        "text": " mundo",
    }


def test_agent_delta_rejects_wrong_type():
    with pytest.raises(ValidationError):
        AgentDelta.model_validate(
            {
                "type": "agent.response",
                "session_id": "session-1",
                "turn_id": 7,
                "text": "Olá",
            }
        )