101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
import sys
|
|
|
|
import mock
|
|
import pytest
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.actuation_receiver import ActuationReceiver
|
|
|
|
|
|
@pytest.fixture
|
|
def zmq_context():
|
|
"""
|
|
A pytest fixture that creates and yields a ZMQ context.
|
|
|
|
:return: An initialized ZeroMQ context.
|
|
:rtype: zmq.Context
|
|
"""
|
|
context = zmq.Context()
|
|
yield context
|
|
|
|
|
|
def test_handle_unimplemented_endpoint(zmq_context):
|
|
"""
|
|
Tests that the ``ActuationReceiver.handle_message`` method can
|
|
handle an unknown or unimplemented endpoint without raising an error.
|
|
"""
|
|
receiver = ActuationReceiver(zmq_context)
|
|
# Should not error
|
|
receiver.handle_message({
|
|
"endpoint": "some_endpoint_that_definitely_does_not_exist",
|
|
"data": None,
|
|
})
|
|
|
|
|
|
def test_speech_message_no_data(zmq_context, mocker):
|
|
"""
|
|
Tests that the message handler logs a warning when a speech actuation
|
|
request (`actuate/speech`) is received but contains empty string data.
|
|
"""
|
|
mock_warn = mocker.patch("logging.warn")
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver.handle_message({"endpoint": "actuate/speech", "data": ""})
|
|
|
|
mock_warn.assert_called_with(mock.ANY)
|
|
|
|
|
|
def test_speech_message_invalid_data(zmq_context, mocker):
|
|
"""
|
|
Tests that the message handler logs a warning when a speech actuation
|
|
request (`actuate/speech`) is received with data that is not a string (e.g., a boolean).
|
|
"""
|
|
mock_warn = mocker.patch("logging.warn")
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver.handle_message({"endpoint": "actuate/speech", "data": True})
|
|
|
|
mock_warn.assert_called_with(mock.ANY)
|
|
|
|
|
|
def test_speech_no_qi(zmq_context, mocker):
|
|
"""
|
|
Tests the actuation receiver's behavior when processing a speech request
|
|
but the global state does not have an active QI session.
|
|
"""
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
|
|
mock_qi_session = mock.PropertyMock(return_value=None)
|
|
type(mock_state).qi_session = mock_qi_session
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
|
|
|
|
mock_qi_session.assert_called()
|
|
|
|
|
|
def test_speech(zmq_context, mocker):
|
|
"""
|
|
Tests the core speech actuation functionality by mocking the QI TextToSpeech
|
|
service and verifying that it is called correctly.
|
|
"""
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
|
|
mock_qi = mock.Mock()
|
|
sys.modules["qi"] = mock_qi
|
|
|
|
mock_tts_service = mock.Mock()
|
|
mock_state.qi_session = mock.Mock()
|
|
mock_state.qi_session.service.return_value = mock_tts_service
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._tts_service = None
|
|
receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
|
|
|
|
mock_state.qi_session.service.assert_called_once_with("ALTextToSpeech")
|
|
|
|
mock_qi.async.assert_called_once()
|
|
call_args = mock_qi.async.call_args[0]
|
|
assert call_args[0] == mock_tts_service.say
|
|
assert call_args[1] == "Some message to speak."
|