pepperplus/pepperplus-ri
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

chore: add documentation RI

Browse Source 
Code functionality left unchanged, only added docs where missing

close: N25B-298
This commit is contained in:
Pim Hutting
2025-11-21 16:35:40 +01:00
parent 1e3531ac6e
commit 051f904576
18 changed files with 629 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
test/common/microphone_utils.py
View File

@@ -17,10 +17,15 @@ class MicrophoneUtils(object):
def test_choose_mic_default(self, pyaudio_instance):
"""
The result must contain at least "index", as this is used to identify the microphone.
The "name" is used for logging, so it should also exist.
It must have one or more channels.
Lastly it must be capable of sending at least 16000 samples per second.
Tests that the default microphone selection function returns a valid
microphone dictionary containing all necessary keys with correct types and values.
The result must contain at least "index", as this is used to identify the microphone,
and "name" for logging. It must have one or more channels (`maxInputChannels`),
and a default sample rate of at least 16000 Hz.
:param pyaudio_instance: A mocked or real PyAudio instance used to query microphone information.
:type pyaudio_instance: PyAudio
"""
result = choose_mic_default(pyaudio_instance)
assert "index" in result
@@ -39,7 +44,15 @@ class MicrophoneUtils(object):
def test_choose_mic_interactive_input_not_int(self, pyaudio_instance, mocker):
"""
First mock an input that's not an integer, then a valid integer. There should be no errors.
Tests the robustness of the interactive selection when the user first enters
a non-integer value, ensuring the system prompts again without error and accepts
a valid integer on the second attempt.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
microphones = get_microphones(pyaudio_instance)
target_microphone = next(microphones)
@@ -59,7 +72,14 @@ class MicrophoneUtils(object):
def test_choose_mic_interactive_negative_index(self, pyaudio_instance, mocker):
"""
Make sure that the interactive method does not allow negative integers as input.
Tests that the interactive selection method prevents the user from entering
a negative integer as a microphone index.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
microphones = get_microphones(pyaudio_instance)
target_microphone = next(microphones)
@@ -79,7 +99,14 @@ class MicrophoneUtils(object):
def test_choose_mic_interactive_index_too_high(self, pyaudio_instance, mocker):
"""
Make sure that the interactive method does not allow indices higher than the highest mic index.
Tests that the interactive selection method prevents the user from entering
an index that exceeds the total number of available microphones.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
real_count = len(list(get_microphones(pyaudio_instance)))
mock_input = mocker.patch("__builtin__.raw_input", side_effect=[str(real_count), "0"])
@@ -96,7 +123,15 @@ class MicrophoneUtils(object):
def test_choose_mic_interactive_random_index(self, pyaudio_instance, mocker):
"""
Get a random index from the list of available mics, make sure it's correct.
Tests the core interactive functionality by simulating the selection of a
random valid microphone index and verifying that the correct microphone
information is returned.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
microphones = list(get_microphones(pyaudio_instance))
random_index = random.randrange(len(microphones))
@@ -115,6 +150,16 @@ class MicrophoneUtils(object):
assert result is None
def test_choose_mic_arguments(self, pyaudio_instance, mocker):
"""
Tests `choose_mic_arguments` when the microphone name is passed as a separate
argument.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
for mic in get_microphones(pyaudio_instance):
mocker.patch.object(sys, "argv", ["--microphone", mic["name"]])
@@ -124,6 +169,16 @@ class MicrophoneUtils(object):
assert result == mic
def test_choose_mic_arguments_eq(self, pyaudio_instance, mocker):
"""
Tests `choose_mic_arguments` when the microphone name is passed using an
equals sign (`--microphone=NAME`).
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
for mic in get_microphones(pyaudio_instance):
mocker.patch.object(sys, "argv", ["--microphone={}".format(mic["name"])])
@@ -132,7 +187,17 @@ class MicrophoneUtils(object):
assert result is not None
assert result == mic
def test_choose_mic_arguments_not_exits(self, pyaudio_instance, mocker):
def test_choose_mic_arguments_not_exist(self, pyaudio_instance, mocker):
"""
Tests `choose_mic_arguments` when a non-existent microphone name is passed
via command-line arguments, expecting the function to return None.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mocker.patch.object(sys, "argv", ["--microphone", "Surely this microphone doesn't exist"])
result = choose_mic_arguments(pyaudio_instance)
@@ -140,6 +205,16 @@ class MicrophoneUtils(object):
assert result is None
def test_choose_mic_with_argument(self, pyaudio_instance, mocker):
"""
Tests `choose_mic` function when a valid microphone is
specified via command-line arguments.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mic = next(get_microphones(pyaudio_instance))
mocker.patch.object(sys, "argv", ["--microphone", mic["name"]])
@@ -149,6 +224,17 @@ class MicrophoneUtils(object):
assert result == mic
def test_choose_mic_no_argument(self, pyaudio_instance, mocker):
"""
Tests `choose_mic` function when no command-line arguments
are provided, verifying that the function falls back correctly to the
system's default microphone selection.
:param pyaudio_instance: A mocked or real PyAudio instance.
:type pyaudio_instance: PyAudio
:param mocker: The fixture used for mocking built-in functions and system objects.
:type mocker: pytest_mock.plugin.MockerFixture
"""
default_mic = choose_mic_default(pyaudio_instance)
mocker.patch.object(sys, "argv", [])

11
test/integration/test_microphone_utils.py
View File

@@ -7,6 +7,17 @@ from common.microphone_utils import MicrophoneUtils
@pytest.fixture
def pyaudio_instance():
"""
A pytest fixture that provides an initialized PyAudio instance for tests
requiring microphone access.
It first initializes PyAudio. If a default input device (microphone) is not
found, the test is skipped to avoid failures in environments
without a mic.
:return: An initialized PyAudio instance.
:rtype: pyaudio.PyAudio
"""
audio = pyaudio.PyAudio()
try:
audio.get_default_input_device_info()

53
test/unit/test_actuation_receiver.py
View File

@@ -9,11 +9,24 @@ from robot_interface.endpoints.actuation_receiver import ActuationReceiver
@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.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = ActuationReceiver(zmq_context)
# Should not error
receiver.handle_message({
@@ -23,6 +36,16 @@ def test_handle_unimplemented_endpoint(zmq_context):
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.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch `logging.warn`.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_warn = mocker.patch("logging.warn")
receiver = ActuationReceiver(zmq_context)
@@ -32,6 +55,16 @@ def test_speech_message_no_data(zmq_context, mocker):
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).
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch `logging.warn`.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_warn = mocker.patch("logging.warn")
receiver = ActuationReceiver(zmq_context)
@@ -41,6 +74,16 @@ def test_speech_message_invalid_data(zmq_context, mocker):
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.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch the global state.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_qi_session = mock.PropertyMock(return_value=None)
@@ -53,6 +96,16 @@ def test_speech_no_qi(zmq_context, mocker):
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.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch state and modules.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_qi = mock.Mock()

52
test/unit/test_audio_sender.py
View File

@@ -11,11 +11,26 @@ 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.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch internal dependencies.
:type mocker: pytest_mock.plugin.MockerFixture
"""
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
@@ -31,6 +46,16 @@ def test_no_microphone(zmq_context, mocker):
def test_unicode_mic_name(zmq_context, mocker):
"""
Tests the robustness of the `AudioSender` when handling microphone names
that contain Unicode characters.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
:param mocker: The pytest-mock fixture used to patch internal dependencies.
:type mocker: pytest_mock.plugin.MockerFixture
"""
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"}
@@ -47,10 +72,25 @@ def test_unicode_mic_name(zmq_context, mocker):
def _fake_read(num_frames):
"""
Helper function to simulate reading raw audio data from a microphone stream.
:param num_frames: The number of audio frames requested.
:type num_frames: int
:return: A byte string containing random data, simulating audio.
:rtype: str
"""
return os.urandom(num_frames * 4)
def test_sending_audio(mocker):
"""
Tests the successful sending of audio data over a ZeroMQ socket.
:param mocker: The pytest-mock fixture used to patch internal dependencies.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
@@ -76,10 +116,22 @@ def test_sending_audio(mocker):
def _fake_read_error(num_frames):
"""
Helper function to simulate an I/O error during microphone stream reading.
:param num_frames: The number of audio frames requested.
:type num_frames: int
"""
raise IOError()
def test_break_microphone(mocker):
"""
Tests the error handling when the microphone stream breaks (raises an IOError).
:param mocker: The pytest-mock fixture used to patch internal dependencies.
:type mocker: pytest_mock.plugin.MockerFixture
"""
mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}

45
test/unit/test_main_receiver.py
View File

@@ -7,11 +7,23 @@ from robot_interface.endpoints.main_receiver import MainReceiver
@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_ping(zmq_context):
"""
Tests the receiver's ability to handle the "ping" endpoint with data.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = MainReceiver(zmq_context)
response = receiver.handle_message({"endpoint": "ping", "data": "pong"})
@@ -22,6 +34,13 @@ def test_handle_ping(zmq_context):
def test_handle_ping_none(zmq_context):
"""
Tests the receiver's ability to handle the ping endpoint when the
data field is explicitly set to None.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = MainReceiver(zmq_context)
response = receiver.handle_message({"endpoint": "ping", "data": None})
@@ -33,6 +52,15 @@ def test_handle_ping_none(zmq_context):
@mock.patch("robot_interface.endpoints.main_receiver.state")
def test_handle_negotiate_ports(mock_state, zmq_context):
"""
Tests the handling of the "negotiate/ports" endpoint.
:param mock_state: Mocked global application state.
:type mock_state: mock.Mock
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = MainReceiver(zmq_context)
mock_state.sockets = [receiver]
@@ -54,6 +82,13 @@ def test_handle_negotiate_ports(mock_state, zmq_context):
def test_handle_unimplemented_endpoint(zmq_context):
"""
Tests that the receiver correctly handles a request to a completely
unknown or non-existent endpoint.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = MainReceiver(zmq_context)
response = receiver.handle_message({
"endpoint": "some_endpoint_that_definitely_does_not_exist",
@@ -67,6 +102,16 @@ def test_handle_unimplemented_endpoint(zmq_context):
def test_handle_unimplemented_negotiation_endpoint(zmq_context):
"""
Tests handling a request to an unknown sub-endpoint within a known
group
The expected behavior is to return a specific "negotiate/error" response
with a descriptive error string.
:param zmq_context: The ZeroMQ context fixture.
:type zmq_context: zmq.Context
"""
receiver = MainReceiver(zmq_context)
response = receiver.handle_message({
"endpoint": "negotiate/but_some_subpath_that_definitely_does_not_exist",

55
test/unit/test_microphone_utils.py
View File

@@ -7,6 +7,16 @@ from robot_interface.utils.microphone import choose_mic_default, choose_mic_inte
class MockPyAudio:
"""
A mock implementation of the PyAudio library class, designed for testing
microphone utility functions without requiring actual audio hardware.
It provides fake devices, including one input microphone, and implements
the core PyAudio methods required for device enumeration.
:ivar devices: A list of dictionaries representing mock audio devices.
:type devices: List[Dict[str, Any]]
"""
def __init__(self):
# You can predefine fake device info here
self.devices = [
@@ -37,18 +47,36 @@ class MockPyAudio:
]
def get_device_count(self):
"""Return the number of available mock devices."""
"""
Returns the number of available mock devices.
:return: The total number of devices in the mock list.
:rtype: int
"""
return len(self.devices)
def get_device_info_by_index(self, index):
"""Return information for a given mock device index."""
"""
Returns information for a given mock device index.
:param index: The index of the device to retrieve.
:type index: int
:return: A dictionary containing device information.
:rtype: Dict[str, Any]
"""
if 0 <= index < len(self.devices):
return self.devices[index]
else:
raise IOError("Invalid device index: {}".format(index))
def get_default_input_device_info(self):
"""Return info for a default mock input device."""
"""
Returns information for the default mock input device.
:return: A dictionary containing the default input device information.
:rtype: Dict[str, Any]
"""
for device in self.devices:
if device.get("maxInputChannels", 0) > 0:
return device
@@ -57,16 +85,32 @@ class MockPyAudio:
@pytest.fixture
def pyaudio_instance():
"""
A pytest fixture that returns an instance of the `MockPyAudio` class.
:return: An initialized instance of the mock PyAudio class.
:rtype: MockPyAudio
"""
return MockPyAudio()
def _raise_io_error():
"""
Helper function used to mock PyAudio methods that are expected to fail
when no device is available.
"""
raise IOError()
class TestAudioUnit(MicrophoneUtils):
"""Run shared audio behavior tests with the mock implementation."""
"""
Runs the shared microphone behavior tests defined in `MicrophoneUtils` using
the mock PyAudio implementation.
"""
def test_choose_mic_default_no_mic(self):
"""
Tests `choose_mic_default` when no microphones are available.
"""
mock_pyaudio = mock.Mock()
mock_pyaudio.get_device_count = mock.Mock(return_value=0L)
mock_pyaudio.get_default_input_device_info = _raise_io_error
@@ -76,6 +120,9 @@ class TestAudioUnit(MicrophoneUtils):
assert result is None
def test_choose_mic_interactive_no_mic(self):
"""
Tests `choose_mic_interactive` when no microphones are available.
"""
mock_pyaudio = mock.Mock()
mock_pyaudio.get_device_count = mock.Mock(return_value=0L)
mock_pyaudio.get_default_input_device_info = _raise_io_error

18
test/unit/test_time_block.py
View File

@@ -6,11 +6,21 @@ from robot_interface.utils.timeblock import TimeBlock
class AnyFloat(object):
"""
A helper class used in tests to assert that a mock function was called
with an argument that is specifically a float, regardless of its value.
It overrides the equality comparison (`__eq__`) to check only the type.
"""
def __eq__(self, other):
return isinstance(other, float)
def test_no_limit():
"""
Tests the scenario where the `TimeBlock` context manager is used without
a time limit.
"""
callback = mock.Mock()
with TimeBlock(callback):
@@ -20,6 +30,10 @@ def test_no_limit():
def test_exceed_limit():
"""
Tests the scenario where the execution time within the `TimeBlock`
exceeds the provided limit.
"""
callback = mock.Mock()
with TimeBlock(callback, 0):
@@ -29,6 +43,10 @@ def test_exceed_limit():
def test_within_limit():
"""
Tests the scenario where the execution time within the `TimeBlock`
stays within the provided limit.
"""
callback = mock.Mock()
with TimeBlock(callback, 5):