182 lines
3.5 KiB
Python
182 lines
3.5 KiB
Python
import pytest
|
|
|
|
from voice_assistant.turn_manager import (
|
|
TurnManager,
|
|
TurnState,
|
|
)
|
|
|
|
|
|
def test_start_turn():
|
|
manager = TurnManager()
|
|
|
|
turn = manager.start(1)
|
|
|
|
assert turn.turn_id == 1
|
|
assert turn.state == TurnState.LISTENING
|
|
assert turn.partial_text == ""
|
|
assert turn.final_text is None
|
|
|
|
|
|
def test_partial_replaces_previous_partial():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
manager.on_partial(1, "Ô Alex")
|
|
assert manager.active_turn is not None
|
|
assert manager.active_turn.partial_text == "Ô Alex"
|
|
|
|
manager.on_partial(1, "Ô Alex tudo bem")
|
|
assert manager.active_turn.partial_text == "Ô Alex tudo bem"
|
|
|
|
|
|
def test_final_moves_turn_to_processing():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
manager.on_partial(1, "Ô Alex tudo bem")
|
|
manager.on_final(1, "Ô Alex tudo bem?")
|
|
|
|
assert manager.active_turn is not None
|
|
assert manager.active_turn.final_text == "Ô Alex tudo bem?"
|
|
assert manager.active_turn.partial_text == "Ô Alex tudo bem?"
|
|
assert manager.active_turn.state == TurnState.PROCESSING
|
|
|
|
|
|
def test_cannot_start_second_turn():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="A turn is already active",
|
|
):
|
|
manager.start(2)
|
|
|
|
|
|
def test_interrupt():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
manager.interrupt(1)
|
|
|
|
assert manager.active_turn is not None
|
|
assert (
|
|
manager.active_turn.state
|
|
== TurnState.INTERRUPTED
|
|
)
|
|
|
|
|
|
def test_clear_turn():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
manager.clear(1)
|
|
|
|
assert manager.active_turn is None
|
|
|
|
|
|
def test_agent_response_moves_turn_to_speaking():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
manager.on_final(
|
|
1,
|
|
"Qual o horário?",
|
|
)
|
|
|
|
manager.on_agent_response(1)
|
|
|
|
assert manager.active_turn is not None
|
|
assert (
|
|
manager.active_turn.state
|
|
== TurnState.SPEAKING
|
|
)
|
|
|
|
|
|
def test_tts_completion_moves_turn_to_completed():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
manager.on_final(
|
|
1,
|
|
"Qual o horário?",
|
|
)
|
|
|
|
manager.on_agent_response(1)
|
|
manager.on_tts_complete(1)
|
|
|
|
assert manager.active_turn is not None
|
|
assert (
|
|
manager.active_turn.state
|
|
== TurnState.COMPLETED
|
|
)
|
|
|
|
|
|
def test_agent_response_requires_processing_state():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="turn is listening",
|
|
):
|
|
manager.on_agent_response(1)
|
|
|
|
|
|
def test_tts_completion_requires_speaking_state():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="turn is listening",
|
|
):
|
|
manager.on_tts_complete(1)
|
|
|
|
|
|
def test_cannot_stop_while_listening():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="Cannot stop turn while turn is listening",
|
|
):
|
|
manager.stop(1)
|
|
|
|
|
|
def test_clear_removes_completed_turn():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
manager.on_final(
|
|
1,
|
|
"Olá!",
|
|
)
|
|
|
|
manager.on_agent_response(1)
|
|
manager.on_tts_complete(1)
|
|
|
|
manager.clear(1)
|
|
|
|
assert manager.active_turn is None
|
|
|
|
|
|
def test_clear_rejects_unexpected_turn_id():
|
|
manager = TurnManager()
|
|
|
|
manager.start(1)
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Unexpected turn_id=2",
|
|
):
|
|
manager.clear(2) |