56c804b7eb
Exhaustive test cases for both classes, with 100% coverage. Adds `mock` dependency. Tests for actuation receiver do not yet pass. ref: N25B-168
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
import sys
|
|
import unittest
|
|
|
|
import mock
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.actuation_receiver import ActuationReceiver
|
|
|
|
|
|
class TestActuationReceiver(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.context = zmq.Context()
|
|
|
|
def test_handle_unimplemented_endpoint(self):
|
|
receiver = ActuationReceiver(self.context)
|
|
# Should not error
|
|
receiver.handle_message({
|
|
"endpoint": "some_endpoint_that_definitely_does_not_exist",
|
|
"data": None,
|
|
})
|
|
|
|
@mock.patch("logging.warn")
|
|
def test_speech_message_no_data(self, mock_warn):
|
|
receiver = ActuationReceiver(self.context)
|
|
receiver.handle_message({"endpoint": "actuate/speech", "data": ""})
|
|
|
|
mock_warn.assert_called_with(mock.ANY)
|
|
|
|
@mock.patch("logging.warn")
|
|
def test_speech_message_invalid_data(self, mock_warn):
|
|
receiver = ActuationReceiver(self.context)
|
|
receiver.handle_message({"endpoint": "actuate/speech", "data": True})
|
|
|
|
mock_warn.assert_called_with(mock.ANY)
|
|
|
|
@mock.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
def test_speech_no_qi(self, mock_state):
|
|
mock_qi_session = mock.PropertyMock(return_value=None)
|
|
type(mock_state).qi_session = mock_qi_session
|
|
|
|
receiver = ActuationReceiver(self.context)
|
|
receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
|
|
|
|
mock_qi_session.assert_called()
|
|
|
|
@mock.patch("robot_interface.endpoints.actuation_receiver.state")
|
|
def test_speech(self, mock_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(self.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]
|
|
self.assertEqual(call_args[0], mock_tts_service.say)
|
|
self.assertEqual(call_args[1], "Some message to speak.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|