Merge branch 'fix/correct-vad-starting' into 'dev'

Move VAD agent creation to RI communication agent

See merge request ics/sp/2025/n25b/pepperplus-cb!34

test/integration/agents/perception/vad_agent/test_vad_agent.py

@@ -5,6 +5,7 @@ import pytest
 import zmq
 
 from control_backend.agents.perception.vad_agent import VADAgent
+from control_backend.schemas.program_status import PROGRAM_STATUS, ProgramStatus
 
 
 @pytest.fixture
@@ -43,14 +44,12 @@ async def test_normal_setup(per_transcription_agent):
         coro.close()
 
     per_vad_agent.add_behavior = swallow_background_task
-    per_vad_agent.reset_stream = AsyncMock()
 
     await per_vad_agent.setup()
 
     per_transcription_agent.assert_called_once()
     per_transcription_agent.return_value.start.assert_called_once()
     per_vad_agent._streaming_loop.assert_called_once()
-    per_vad_agent.reset_stream.assert_called_once()
     assert per_vad_agent.audio_in_socket is not None
     assert per_vad_agent.audio_out_socket is not None
 
@@ -103,7 +102,7 @@ async def test_out_socket_creation_failure(zmq_context):
     zmq_context.return_value.socket.return_value.bind_to_random_port.side_effect = zmq.ZMQBindError
     per_vad_agent = VADAgent("tcp://localhost:12345", False)
     per_vad_agent.stop = AsyncMock()
-    per_vad_agent.reset_stream = AsyncMock()
+    per_vad_agent._reset_stream = AsyncMock()
     per_vad_agent._streaming_loop = AsyncMock()
     per_vad_agent._connect_audio_out_socket = MagicMock(return_value=None)
 
@@ -124,7 +123,7 @@ async def test_stop(zmq_context, per_transcription_agent):
     Test that when the VAD agent is stopped, the sockets are closed correctly.
     """
     per_vad_agent = VADAgent("tcp://localhost:12345", False)
-    per_vad_agent.reset_stream = AsyncMock()
+    per_vad_agent._reset_stream = AsyncMock()
     per_vad_agent._streaming_loop = AsyncMock()
 
     async def swallow_background_task(coro):
@@ -142,3 +141,66 @@ async def test_stop(zmq_context, per_transcription_agent):
     assert zmq_context.return_value.socket.return_value.close.call_count == 2
     assert per_vad_agent.audio_in_socket is None
     assert per_vad_agent.audio_out_socket is None
+
+
+@pytest.mark.asyncio
+async def test_application_startup_complete(zmq_context):
+    """Check that it resets the stream when the program finishes startup."""
+    vad_agent = VADAgent("tcp://localhost:12345", False)
+    vad_agent._running = True
+    vad_agent._reset_stream = AsyncMock()
+    vad_agent.program_sub_socket = AsyncMock()
+    vad_agent.program_sub_socket.recv_multipart.side_effect = [
+        (PROGRAM_STATUS, ProgramStatus.RUNNING.value),
+    ]
+
+    await vad_agent._status_loop()
+
+    vad_agent._reset_stream.assert_called_once()
+    vad_agent.program_sub_socket.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_application_other_status(zmq_context):
+    """
+    Check that it does nothing when the internal communication message is a status update, but not
+    running.
+    """
+    vad_agent = VADAgent("tcp://localhost:12345", False)
+    vad_agent._running = True
+    vad_agent._reset_stream = AsyncMock()
+    vad_agent.program_sub_socket = AsyncMock()
+
+    vad_agent.program_sub_socket.recv_multipart.side_effect = [
+        (PROGRAM_STATUS, ProgramStatus.STARTING.value),
+        (PROGRAM_STATUS, ProgramStatus.STOPPING.value),
+    ]
+    try:
+        # Raises StopAsyncIteration the third time it calls `program_sub_socket.recv_multipart`
+        await vad_agent._status_loop()
+    except StopAsyncIteration:
+        pass
+
+    vad_agent._reset_stream.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_application_message_other(zmq_context):
+    """
+    Check that it does nothing when there's an internal communication message that is not a status
+    update.
+    """
+    vad_agent = VADAgent("tcp://localhost:12345", False)
+    vad_agent._running = True
+    vad_agent._reset_stream = AsyncMock()
+    vad_agent.program_sub_socket = AsyncMock()
+
+    vad_agent.program_sub_socket.recv_multipart.side_effect = [(b"internal/other", b"Whatever")]
+
+    try:
+        # Raises StopAsyncIteration the second time it calls `program_sub_socket.recv_multipart`
+        await vad_agent._status_loop()
+    except StopAsyncIteration:
+        pass
+
+    vad_agent._reset_stream.assert_not_called()