From 5631a55697a815b102e239bef0f5bdf0f40ea5ca Mon Sep 17 00:00:00 2001 From: Twirre Meulenbelt <43213592+TwirreM@users.noreply.github.com> Date: Tue, 21 Oct 2025 13:55:06 +0200 Subject: [PATCH] test: convert to pytest Instead of built-in `unittest`, now use `pytest`. Find versions that work, convert tests. ref: N25B-168 --- README.md | 8 +- requirements.txt | 4 +- test/unit/test_actuation_receiver.py | 123 ++++++++++++----------- test/unit/test_main_receiver.py | 142 +++++++++++++-------------- test/unit/test_time_block.py | 47 +++++---- 5 files changed, 166 insertions(+), 158 deletions(-) diff --git a/README.md b/README.md index 6fffab0..5a10267 100644 --- a/README.md +++ b/README.md @@ -105,11 +105,15 @@ With both, if you want to connect to the actual robot (or simulator), pass the ` To run the unit tests, on Linux and macOS: ```shell -PYTHONPATH=src python -m unittest discover -s test -p "test_*.py" -v +PYTHONPATH=src pytest test/ ``` On Windows: ```shell -$env:PYTHONPATH="src"; python -m unittest discover -s test -p "test_*.py" -v +$env:PYTHONPATH="src"; pytest test/ ``` + +### Coverage + +For coverage, add `--cov=robot_interface` as an argument to `pytest`. diff --git a/requirements.txt b/requirements.txt index 84b4d20..f93c70d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ pyzmq<16 pyaudio<=0.2.11 -mock~=3.0.5 +pytest<5 +pytest-mock<3.0.0 +pytest-cov<3.0.0 diff --git a/test/unit/test_actuation_receiver.py b/test/unit/test_actuation_receiver.py index 11039e6..da70964 100644 --- a/test/unit/test_actuation_receiver.py +++ b/test/unit/test_actuation_receiver.py @@ -1,69 +1,74 @@ import sys -import unittest import mock +import pytest 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.") +@pytest.fixture +def zmq_context(): + context = zmq.Context() + yield context -if __name__ == "__main__": - unittest.main() +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." diff --git a/test/unit/test_main_receiver.py b/test/unit/test_main_receiver.py index 31c34cc..4ded502 100644 --- a/test/unit/test_main_receiver.py +++ b/test/unit/test_main_receiver.py @@ -1,79 +1,79 @@ -import unittest - import mock +import pytest import zmq from robot_interface.endpoints.main_receiver import MainReceiver -class TestMainReceiver(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.context = zmq.Context() - - def test_handle_ping(self): - receiver = MainReceiver(self.context) - response = receiver.handle_message({"endpoint": "ping", "data": "pong"}) - - self.assertIn("endpoint", response) - self.assertEqual(response["endpoint"], "ping") - self.assertIn("data", response) - self.assertEqual(response["data"], "pong") - - def test_handle_ping_none(self): - receiver = MainReceiver(self.context) - response = receiver.handle_message({"endpoint": "ping", "data": None}) - - self.assertIn("endpoint", response) - self.assertEqual(response["endpoint"], "ping") - self.assertIn("data", response) - self.assertEqual(response["data"], None) - - @mock.patch("robot_interface.endpoints.main_receiver.state") - def test_handle_negotiate_ports(self, mock_state): - receiver = MainReceiver(self.context) - mock_state.sockets = [receiver] - - response = receiver.handle_message({"endpoint": "negotiate/ports", "data": None}) - - self.assertIn("endpoint", response) - self.assertEqual(response["endpoint"], "negotiate/ports") - self.assertIn("data", response) - self.assertIsInstance(response["data"], list) - for port in response["data"]: - self.assertIn("id", port) - self.assertIsInstance(port["id"], str) - self.assertIn("port", port) - self.assertIsInstance(port["port"], int) - self.assertIn("bind", port) - self.assertIsInstance(port["bind"], bool) - - self.assertTrue(any(port["id"] == "main" for port in response["data"])) - - def test_handle_unimplemented_endpoint(self): - receiver = MainReceiver(self.context) - response = receiver.handle_message({ - "endpoint": "some_endpoint_that_definitely_does_not_exist", - "data": None, - }) - - self.assertIn("endpoint", response) - self.assertEqual(response["endpoint"], "error") - self.assertIn("data", response) - self.assertIsInstance(response["data"], str) - - def test_handle_unimplemented_negotiation_endpoint(self): - receiver = MainReceiver(self.context) - response = receiver.handle_message({ - "endpoint": "negotiate/but_some_subpath_that_definitely_does_not_exist", - "data": None, - }) - - self.assertIn("endpoint", response) - self.assertEqual(response["endpoint"], "negotiate/error") - self.assertIn("data", response) - self.assertIsInstance(response["data"], str) +@pytest.fixture +def zmq_context(): + context = zmq.Context() + yield context -if __name__ == "__main__": - unittest.main() +def test_handle_ping(zmq_context): + receiver = MainReceiver(zmq_context) + response = receiver.handle_message({"endpoint": "ping", "data": "pong"}) + + assert "endpoint" in response + assert response["endpoint"] == "ping" + assert "data" in response + assert response["data"] == "pong" + + +def test_handle_ping_none(zmq_context): + receiver = MainReceiver(zmq_context) + response = receiver.handle_message({"endpoint": "ping", "data": None}) + + assert "endpoint" in response + assert response["endpoint"] == "ping" + assert "data" in response + assert response["data"] == None + + +@mock.patch("robot_interface.endpoints.main_receiver.state") +def test_handle_negotiate_ports(mock_state, zmq_context): + receiver = MainReceiver(zmq_context) + mock_state.sockets = [receiver] + + response = receiver.handle_message({"endpoint": "negotiate/ports", "data": None}) + + assert "endpoint" in response + assert response["endpoint"] == "negotiate/ports" + assert "data" in response + assert isinstance(response["data"], list) + for port in response["data"]: + assert "id" in port + assert isinstance(port["id"], str) + assert "port" in port + assert isinstance(port["port"], int) + assert "bind" in port + assert isinstance(port["bind"], bool) + + assert any(port["id"] == "main" for port in response["data"]) + + +def test_handle_unimplemented_endpoint(zmq_context): + receiver = MainReceiver(zmq_context) + response = receiver.handle_message({ + "endpoint": "some_endpoint_that_definitely_does_not_exist", + "data": None, + }) + + assert "endpoint" in response + assert response["endpoint"] == "error" + assert "data" in response + assert isinstance(response["data"], str) + + +def test_handle_unimplemented_negotiation_endpoint(zmq_context): + receiver = MainReceiver(zmq_context) + response = receiver.handle_message({ + "endpoint": "negotiate/but_some_subpath_that_definitely_does_not_exist", + "data": None, + }) + + assert "endpoint" in response + assert response["endpoint"] == "negotiate/error" + assert "data" in response + assert isinstance(response["data"], str) diff --git a/test/unit/test_time_block.py b/test/unit/test_time_block.py index b6cabbc..eabc91b 100644 --- a/test/unit/test_time_block.py +++ b/test/unit/test_time_block.py @@ -1,4 +1,4 @@ -import unittest +import time import mock @@ -10,31 +10,28 @@ class AnyFloat(object): return isinstance(other, float) -class TestTimeBlock(unittest.TestCase): - def test_no_limit(self): - callback = mock.Mock() +def test_no_limit(): + callback = mock.Mock() - with TimeBlock(callback): - pass + with TimeBlock(callback): + pass - callback.assert_called_once_with(AnyFloat()) - - def test_exceed_limit(self): - callback = mock.Mock() - - with TimeBlock(callback, 0): - pass - - callback.assert_called_once_with(AnyFloat()) - - def test_within_limit(self): - callback = mock.Mock() - - with TimeBlock(callback, 5): - pass - - callback.assert_not_called() + callback.assert_called_once_with(AnyFloat()) -if __name__ == '__main__': - unittest.main() +def test_exceed_limit(): + callback = mock.Mock() + + with TimeBlock(callback, 0): + time.sleep(0.001) + + callback.assert_called_once_with(AnyFloat()) + + +def test_within_limit(): + callback = mock.Mock() + + with TimeBlock(callback, 5): + pass + + callback.assert_not_called()