105 lines
3.0 KiB
Python
105 lines
3.0 KiB
Python
# coding=utf-8
|
|
import os
|
|
import time
|
|
|
|
import mock
|
|
import pytest
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.audio_sender import AudioSender
|
|
|
|
|
|
@pytest.fixture
|
|
def zmq_context():
|
|
context = zmq.Context()
|
|
yield context
|
|
|
|
|
|
def test_no_microphone(zmq_context, mocker):
|
|
mock_info_logger = mocker.patch("robot_interface.endpoints.audio_sender.logger.info")
|
|
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
|
|
mock_choose_mic.return_value = None
|
|
|
|
sender = AudioSender(zmq_context)
|
|
assert sender.microphone is None
|
|
|
|
sender.start()
|
|
assert sender.thread is None
|
|
mock_info_logger.assert_called()
|
|
|
|
sender.wait_until_done() # Should return early because we didn't start a thread
|
|
|
|
|
|
def test_unicode_mic_name(zmq_context, mocker):
|
|
mocker.patch("robot_interface.endpoints.audio_sender.threading")
|
|
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
|
|
mock_choose_mic.return_value = {"name": u"• Some Unicode name"}
|
|
|
|
sender = AudioSender(zmq_context)
|
|
assert sender.microphone is not None
|
|
|
|
# `.start()` logs the name of the microphone. It should not give an error if it contains Unicode
|
|
# symbols.
|
|
sender.start()
|
|
assert sender.thread is not None
|
|
|
|
sender.wait_until_done() # Should return instantly because we didn't start a real thread
|
|
|
|
|
|
def _fake_read(num_frames):
|
|
return os.urandom(num_frames * 4)
|
|
|
|
|
|
def test_sending_audio(mocker):
|
|
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
|
|
mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
|
|
|
|
mock_state = mocker.patch("robot_interface.endpoints.audio_sender.state")
|
|
mock_state.exit_event.is_set.side_effect = [False, True]
|
|
|
|
mock_zmq_context = mock.Mock()
|
|
send_socket = mock.Mock()
|
|
|
|
# If there's something wrong with the microphone, it will raise an IOError when `read`ing.
|
|
stream = mock.Mock()
|
|
stream.read = _fake_read
|
|
|
|
sender = AudioSender(mock_zmq_context)
|
|
sender.socket.send = send_socket
|
|
sender.audio.open = mock.Mock()
|
|
sender.audio.open.return_value = stream
|
|
|
|
sender.start()
|
|
sender.wait_until_done()
|
|
|
|
send_socket.assert_called()
|
|
|
|
|
|
def _fake_read_error(num_frames):
|
|
raise IOError()
|
|
|
|
|
|
def test_break_microphone(mocker):
|
|
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
|
|
mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
|
|
|
|
mock_state = mocker.patch("robot_interface.endpoints.audio_sender.state")
|
|
mock_state.exit_event.is_set.side_effect = [False, True]
|
|
|
|
mock_zmq_context = mock.Mock()
|
|
send_socket = mock.Mock()
|
|
|
|
# If there's something wrong with the microphone, it will raise an IOError when `read`ing.
|
|
stream = mock.Mock()
|
|
stream.read = _fake_read_error
|
|
|
|
sender = AudioSender(mock_zmq_context)
|
|
sender.socket.send = send_socket
|
|
sender.audio.open = mock.Mock()
|
|
sender.audio.open.return_value = stream
|
|
|
|
sender.start()
|
|
sender.wait_until_done()
|
|
|
|
send_socket.assert_not_called()
|