254 lines
8.9 KiB
Python
254 lines
8.9 KiB
Python
import sys
|
|
|
|
import mock
|
|
import pytest
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.actuation_receiver import ActuationReceiver
|
|
from robot_interface.endpoints.gesture_settings import GestureTags
|
|
|
|
|
|
@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")
|
|
|
|
getattr(mock_qi, "async").assert_called_once()
|
|
call_args = getattr(mock_qi, "async").call_args[0]
|
|
assert call_args[0] == mock_tts_service.say
|
|
assert call_args[1] == "Some message to speak."
|
|
|
|
|
|
def test_gesture_no_data(zmq_context, mocker):
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": ""}, True)
|
|
# Just ensuring no crash
|
|
|
|
|
|
def test_gesture_invalid_data(zmq_context, mocker):
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": 123}, True)
|
|
# No crash expected
|
|
|
|
|
|
def test_gesture_single_not_found(zmq_context, mocker):
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.single_gestures = ["wave", "bow"] # allowed single gestures
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "unknown_gesture"}, True)
|
|
# No crash expected
|
|
|
|
|
|
def test_gesture_tag_not_found(zmq_context, mocker):
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.tags = ["happy", "sad"]
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "not_a_tag"}, False)
|
|
# No crash expected
|
|
|
|
|
|
def test_gesture_no_qi_session(zmq_context, mocker):
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
mock_state.qi_session = None
|
|
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.single_gestures = ["hello"]
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "hello"}, True)
|
|
# No crash, path returns early
|
|
|
|
|
|
def test_gesture_single_success(zmq_context, mocker):
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
mock_qi = mock.Mock()
|
|
sys.modules["qi"] = mock_qi
|
|
|
|
# Setup gesture settings
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.single_gestures = ["wave"]
|
|
|
|
mock_animation_service = mock.Mock()
|
|
mock_state.qi_session = mock.Mock()
|
|
mock_state.qi_session.service.return_value = mock_animation_service
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "wave"}, True)
|
|
|
|
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
|
|
getattr(mock_qi, "async").assert_called_once()
|
|
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.run
|
|
assert getattr(mock_qi, "async").call_args[0][1] == "wave"
|
|
|
|
|
|
def test_gesture_tag_success(zmq_context, mocker):
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
mock_qi = mock.Mock()
|
|
sys.modules["qi"] = mock_qi
|
|
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.tags = ["greeting"]
|
|
|
|
mock_animation_service = mock.Mock()
|
|
mock_state.qi_session = mock.Mock()
|
|
mock_state.qi_session.service.return_value = mock_animation_service
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "greeting"}, False)
|
|
|
|
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
|
|
getattr(mock_qi, "async").assert_called_once()
|
|
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.runTag
|
|
assert getattr(mock_qi, "async").call_args[0][1] == "greeting"
|
|
|
|
|
|
def test_handle_message_all_routes(zmq_context, mocker):
|
|
"""
|
|
Ensures all handle_message endpoint branches route correctly.
|
|
"""
|
|
receiver = ActuationReceiver(zmq_context)
|
|
|
|
mock_speech = mocker.patch.object(receiver, "_handle_speech")
|
|
mock_gesture = mocker.patch.object(receiver, "_handle_gesture")
|
|
|
|
receiver.handle_message({"endpoint": "actuate/speech", "data": "hi"})
|
|
receiver.handle_message({"endpoint": "actuate/gesture/tag", "data": "greeting"})
|
|
receiver.handle_message({"endpoint": "actuate/gesture/single", "data": "wave"})
|
|
|
|
mock_speech.assert_called_once()
|
|
assert mock_gesture.call_count == 2
|
|
|
|
|
|
def test_endpoint_description(zmq_context, mocker):
|
|
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
|
|
mock_tags.tags = ["happy"]
|
|
mock_tags.single_gestures = ["wave"]
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
desc = receiver.endpoint_description()
|
|
|
|
assert "gestures" in desc
|
|
assert desc["gestures"] == ["happy"]
|
|
|
|
assert "single_gestures" in desc
|
|
assert desc["single_gestures"] == ["wave"]
|
|
|
|
|
|
def test_gesture_single_real_gesturetags(zmq_context, mocker):
|
|
"""
|
|
Uses the real GestureTags (no mocking) to ensure the receiver
|
|
references GestureTags.single_gestures correctly.
|
|
"""
|
|
# Ensure qi session exists so we pass the early return
|
|
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
mock_state.qi_session = mock.Mock()
|
|
|
|
# Mock qi.async to avoid real async calls
|
|
mock_qi = mock.Mock()
|
|
sys.modules["qi"] = mock_qi
|
|
|
|
# Mock animation service
|
|
mock_animation_service = mock.Mock()
|
|
mock_state.qi_session.service.return_value = mock_animation_service
|
|
|
|
receiver = ActuationReceiver(zmq_context)
|
|
|
|
# Pick a real gesture from GestureTags.single_gestures
|
|
assert len(GestureTags.single_gestures) > 0, "GestureTags.single_gestures must not be empty"
|
|
gesture = GestureTags.single_gestures[0]
|
|
|
|
receiver._handle_gesture(
|
|
{"endpoint": "actuate/gesture/single", "data": gesture},
|
|
is_single=True,
|
|
)
|
|
|
|
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
|
|
getattr(mock_qi, "async").assert_called_once()
|
|
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.run
|
|
assert getattr(mock_qi, "async").call_args[0][1] == gesture
|
|
|