tes
8bb0ce51b1
GA Realtime API requires create_response:true and interrupt_response:true inside turn_detection — without them the model never auto-generates a reply after VAD detects speech_stopped, so the relay idles out at 30s with no assistant audio. Add both. Client side: when the /device WS drops (Coolify rolling deploy, etc.), Session.run_forever reconnects but the StateMachine kept whatever state it was in — usually LISTENING, which permanently disables the wakeword. Add StateMachine.force_idle() and a Callbacks.on_disconnected() hook; main.py wires it to force the state back to IDLE on every disconnect. Also harden Session's fire-and-forget send_* helpers to swallow ConnectionClosed instead of bubbling up as unhandled task exceptions.
128 lines
3.3 KiB
Python
128 lines
3.3 KiB
Python
from client.state import State, StateMachine
|
|
|
|
|
|
def test_starts_idle_with_wakeword_enabled_and_uplink_disabled():
|
|
sm = StateMachine()
|
|
assert sm.state is State.IDLE
|
|
assert sm.wakeword_enabled
|
|
assert not sm.uplink_enabled
|
|
|
|
|
|
def test_wake_fired_goes_idle_to_wake_pending_and_disables_wakeword():
|
|
sm = StateMachine()
|
|
assert sm.wake_fired() is True
|
|
assert sm.state is State.WAKE_PENDING
|
|
assert not sm.wakeword_enabled
|
|
assert not sm.uplink_enabled
|
|
|
|
|
|
def test_wake_fired_is_idempotent_when_not_idle():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
assert sm.wake_fired() is False
|
|
assert sm.state is State.WAKE_PENDING
|
|
|
|
|
|
def test_session_started_goes_wake_pending_to_listening_and_unmutes_uplink():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
assert sm.state is State.LISTENING
|
|
assert sm.uplink_enabled
|
|
assert not sm.wakeword_enabled
|
|
|
|
|
|
def test_assistant_audio_arrived_mutes_uplink():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
sm.assistant_audio_arrived()
|
|
assert sm.state is State.ASSISTANT_SPEAKING
|
|
assert not sm.uplink_enabled
|
|
|
|
|
|
def test_assistant_audio_arrived_is_noop_outside_listening():
|
|
sm = StateMachine()
|
|
sm.assistant_audio_arrived()
|
|
assert sm.state is State.IDLE # ignored from IDLE
|
|
sm.wake_fired()
|
|
sm.assistant_audio_arrived()
|
|
assert sm.state is State.WAKE_PENDING # ignored from WAKE_PENDING
|
|
|
|
|
|
def test_assistant_done_returns_to_listening_from_speaking():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
sm.assistant_audio_arrived()
|
|
sm.assistant_done()
|
|
assert sm.state is State.LISTENING
|
|
assert sm.uplink_enabled
|
|
|
|
|
|
def test_assistant_done_is_noop_when_not_speaking():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started() # LISTENING; never went to ASSISTANT_SPEAKING
|
|
sm.assistant_done()
|
|
assert sm.state is State.LISTENING
|
|
|
|
|
|
def test_session_ended_routes_through_idle_pending_then_idle():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
sm.session_ended()
|
|
assert sm.state is State.IDLE_PENDING
|
|
assert not sm.wakeword_enabled
|
|
assert not sm.uplink_enabled
|
|
sm.sleep_played()
|
|
assert sm.state is State.IDLE
|
|
assert sm.wakeword_enabled
|
|
|
|
|
|
def test_session_ended_works_from_assistant_speaking():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
sm.assistant_audio_arrived()
|
|
sm.session_ended()
|
|
assert sm.state is State.IDLE_PENDING
|
|
|
|
|
|
def test_session_ended_from_idle_is_noop():
|
|
sm = StateMachine()
|
|
sm.session_ended()
|
|
assert sm.state is State.IDLE
|
|
|
|
|
|
def test_sleep_played_outside_idle_pending_is_noop():
|
|
sm = StateMachine()
|
|
sm.sleep_played()
|
|
assert sm.state is State.IDLE
|
|
|
|
|
|
def test_force_idle_from_listening_resets_state_and_reenables_wakeword():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
assert sm.state is State.LISTENING
|
|
sm.force_idle()
|
|
assert sm.state is State.IDLE
|
|
assert sm.wakeword_enabled
|
|
|
|
|
|
def test_force_idle_from_assistant_speaking_resets():
|
|
sm = StateMachine()
|
|
sm.wake_fired()
|
|
sm.session_started()
|
|
sm.assistant_audio_arrived()
|
|
sm.force_idle()
|
|
assert sm.state is State.IDLE
|
|
|
|
|
|
def test_force_idle_on_idle_is_noop():
|
|
sm = StateMachine()
|
|
sm.force_idle()
|
|
assert sm.state is State.IDLE
|