pepperplus-ri/test/unit/test_audio_sender.py

# coding=utf-8
import os

import mock
import pytest
import zmq

from robot_interface.endpoints.audio_sender import AudioSender


@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_no_microphone(zmq_context, mocker):
    """
    Tests the scenario where no valid microphone can be chosen for recording.
    """
    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):
    """
    Tests the robustness of the `AudioSender` when handling microphone names
    that contain Unicode characters.
    """
    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):
    """
    Helper function to simulate reading raw audio data from a microphone stream.
    """
    return os.urandom(num_frames * 4)


def test_sending_audio(mocker):
    """
    Tests the successful sending of audio data over a ZeroMQ socket.
    """
    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()
    
    mock_state.is_speaking = False
    # 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 test_no_sending_if_speaking(mocker):
    """
    Tests the successful sending of audio data over a ZeroMQ socket.
    """
    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()
    
    mock_state.is_speaking = True

    # 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_not_called()


def _fake_read_error(num_frames):
    """
    Helper function to simulate an I/O error during microphone stream reading.
    """
    raise IOError()


def test_break_microphone(mocker):
    """
    Tests the error handling when the microphone stream breaks (raises an IOError).
    """
    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()


def test_pyaudio_init_failure(mocker, zmq_context):
    """
    Tests the behavior when PyAudio initialization fails (raises an IOError).
    """
    # Prevent binding the ZMQ socket
    mocker.patch("robot_interface.endpoints.audio_sender.AudioSender.create_socket")

    # Simulate PyAudio() failing
    mocker.patch(
        "robot_interface.endpoints.audio_sender.pyaudio.PyAudio",
        side_effect=IOError("boom")
    )
    
    sender = AudioSender(zmq_context)

    assert sender.audio is None
    assert sender.microphone is None