5631a55697
Instead of built-in `unittest`, now use `pytest`. Find versions that work, convert tests. ref: N25B-168
75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
import sys
|
|
|
|
import mock
|
|
import pytest
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.actuation_receiver import ActuationReceiver
|
|
|
|
|
|
@pytest.fixture
|
|
def zmq_context():
|
|
context = zmq.Context()
|
|
yield context
|
|
|
|
|
|
def test_handle_unimplemented_endpoint(zmq_context):
|
|
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):
|
|
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):
|
|
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):
|
|
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):
|
|
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."
|