"""Unit tests for TurnStateMachine."""
import pytest
import threading
from src.localvoicemode.state.turn_state import TurnStateMachine
class TestTurnStateMachine:
"""Tests for TurnStateMachine."""
@pytest.fixture
def machine(self):
"""Fresh state machine for each test."""
return TurnStateMachine()
# Initial state tests
def test_initial_state_is_idle(self, machine):
"""Default state is IDLE."""
assert machine.current_state.id == "idle"
assert machine.state_name == "idle"
assert machine.is_idle() is True
def test_state_query_methods(self, machine):
"""State query methods work correctly."""
assert machine.is_idle() is True
assert machine.is_listening() is False
assert machine.is_processing() is False
assert machine.is_speaking() is False
# Normal flow transitions
def test_normal_conversation_flow(self, machine):
"""Complete conversation flow: idle -> listening -> processing -> speaking -> idle."""
# Start listening
machine.start_listening()
assert machine.is_listening() is True
# Voice detected, start processing
machine.voice_detected()
assert machine.is_processing() is True
# Start speaking response
machine.start_speaking()
assert machine.is_speaking() is True
# Finish speaking, back to idle
machine.finish_speaking()
assert machine.is_idle() is True
def test_direct_tts_from_idle(self, machine):
"""Can speak directly from idle (no voice input needed)."""
machine.start_speaking()
assert machine.is_speaking() is True
machine.finish_speaking()
assert machine.is_idle() is True
# Barge-in tests
def test_barge_in_during_speaking(self, machine):
"""User can interrupt (barge-in) during speaking."""
machine.start_speaking()
assert machine.is_speaking() is True
machine.barge_in()
assert machine.is_listening() is True
# Cancel tests
def test_cancel_from_listening(self, machine):
"""Cancel returns to idle from listening."""
machine.start_listening()
machine.cancel()
assert machine.is_idle() is True
def test_cancel_from_processing(self, machine):
"""Cancel returns to idle from processing."""
machine.start_listening()
machine.voice_detected()
machine.cancel()
assert machine.is_idle() is True
def test_cancel_from_speaking(self, machine):
"""Cancel returns to idle from speaking."""
machine.start_speaking()
machine.cancel()
assert machine.is_idle() is True
# Can-transition checks
def test_can_listen_only_from_idle(self, machine):
"""can_listen() is True only when idle."""
assert machine.can_listen() is True
machine.start_listening()
assert machine.can_listen() is False
machine.cancel()
assert machine.can_listen() is True
def test_can_speak_from_idle_or_processing(self, machine):
"""can_speak() is True when idle or processing."""
assert machine.can_speak() is True # idle
machine.start_listening()
assert machine.can_speak() is False # listening
machine.voice_detected()
assert machine.can_speak() is True # processing
# Thread safety tests
def test_safe_send_success(self, machine):
"""safe_send returns True on successful transition."""
result = machine.safe_send("start_listening")
assert result is True
assert machine.is_listening() is True
def test_safe_send_invalid_transition(self, machine):
"""safe_send returns False on invalid transition."""
# Can't voice_detected from idle
result = machine.safe_send("voice_detected")
assert result is False
assert machine.is_idle() is True
def test_safe_send_rejects_concurrent(self, machine):
"""safe_send rejects concurrent transitions."""
results = []
barrier = threading.Barrier(2)
def transition():
barrier.wait() # Synchronize start
result = machine.safe_send("start_listening")
results.append(result)
t1 = threading.Thread(target=transition)
t2 = threading.Thread(target=transition)
t1.start()
t2.start()
t1.join()
t2.join()
# One should succeed, one might fail (or both succeed if timing allows)
# Key is that state is consistent
assert machine.state_name in ["idle", "listening"]
def test_multiple_sequential_transitions_thread_safe(self, machine):
"""Multiple threads doing sequential transitions remain consistent."""
errors = []
def do_cycle():
try:
for _ in range(5):
if machine.safe_send("start_listening"):
machine.safe_send("voice_detected")
machine.safe_send("start_speaking")
machine.safe_send("finish_speaking")
except Exception as e:
errors.append(str(e))
threads = [threading.Thread(target=do_cycle) for _ in range(3)]
for t in threads:
t.start()
for t in threads:
t.join()
# No exceptions, state is valid
assert len(errors) == 0
assert machine.state_name in ["idle", "listening", "processing", "speaking"]
# Callback tests
def test_on_enter_listening_callback(self, machine):
"""on_enter_listening callback is called."""
called = []
machine.set_on_enter_listening(lambda: called.append("enter_listening"))
machine.start_listening()
assert "enter_listening" in called
def test_on_exit_listening_callback(self, machine):
"""on_exit_listening callback is called."""
called = []
machine.set_on_exit_listening(lambda: called.append("exit_listening"))
machine.start_listening()
machine.voice_detected()
assert "exit_listening" in called
def test_on_enter_speaking_callback(self, machine):
"""on_enter_speaking callback is called."""
called = []
machine.set_on_enter_speaking(lambda: called.append("enter_speaking"))
machine.start_speaking()
assert "enter_speaking" in called
def test_on_exit_speaking_callback(self, machine):
"""on_exit_speaking callback is called."""
called = []
machine.set_on_exit_speaking(lambda: called.append("exit_speaking"))
machine.start_speaking()
machine.finish_speaking()
assert "exit_speaking" in called
def test_callback_not_called_if_not_set(self, machine):
"""Callbacks don't error if not set."""
# Should not raise even without callbacks
machine.start_listening()
machine.voice_detected()
machine.start_speaking()
machine.finish_speaking()
assert machine.is_idle() is True
def test_all_callbacks_in_sequence(self, machine):
"""All callbacks fire in correct order during full cycle."""
called = []
machine.set_on_enter_listening(lambda: called.append("enter_listening"))
machine.set_on_exit_listening(lambda: called.append("exit_listening"))
machine.set_on_enter_speaking(lambda: called.append("enter_speaking"))
machine.set_on_exit_speaking(lambda: called.append("exit_speaking"))
machine.start_listening()
machine.voice_detected()
machine.start_speaking()
machine.finish_speaking()
assert called == [
"enter_listening",
"exit_listening",
"enter_speaking",
"exit_speaking",
]
class TestTurnStateMachineInvalidTransitions:
"""Tests for invalid transition handling."""
@pytest.fixture
def machine(self):
return TurnStateMachine()
def test_cannot_voice_detected_from_idle(self, machine):
"""Cannot voice_detected when not listening."""
with pytest.raises(Exception):
machine.voice_detected()
def test_cannot_finish_speaking_when_not_speaking(self, machine):
"""Cannot finish_speaking when not speaking."""
with pytest.raises(Exception):
machine.finish_speaking()
def test_cannot_barge_in_when_not_speaking(self, machine):
"""Cannot barge_in when not speaking."""
with pytest.raises(Exception):
machine.barge_in()
def test_cannot_start_listening_when_not_idle(self, machine):
"""Cannot start_listening when not idle."""
machine.start_listening()
with pytest.raises(Exception):
machine.start_listening()
def test_cannot_cancel_from_idle(self, machine):
"""Cannot cancel when already idle."""
with pytest.raises(Exception):
machine.cancel()
class TestTurnStateMachineStateProperty:
"""Tests for state property and introspection."""
@pytest.fixture
def machine(self):
return TurnStateMachine()
def test_state_name_property(self, machine):
"""state_name property reflects current state."""
assert machine.state_name == "idle"
machine.start_listening()
assert machine.state_name == "listening"
machine.voice_detected()
assert machine.state_name == "processing"
machine.start_speaking()
assert machine.state_name == "speaking"
machine.finish_speaking()
assert machine.state_name == "idle"
def test_current_state_id_matches_state_name(self, machine):
"""current_state.id matches state_name."""
assert machine.current_state.id == machine.state_name
machine.start_listening()
assert machine.current_state.id == machine.state_name
def test_has_lock_attribute(self, machine):
"""Machine has thread lock attribute."""
assert hasattr(machine, "_lock")
assert isinstance(machine._lock, type(threading.Lock()))
