From 18a4bde4ca28ad2019c5f234ec13d40607f2e6cf Mon Sep 17 00:00:00 2001 From: JobvAlewijk Date: Thu, 29 Jan 2026 17:51:39 +0100 Subject: [PATCH] test: added tests and docstrings ref: N25B-397 --- test/unit/test_face_detection_sender.py | 170 ++++++++++++++++++++++++ test/unit/test_main.py | 5 + 2 files changed, 175 insertions(+) create mode 100644 test/unit/test_face_detection_sender.py diff --git a/test/unit/test_face_detection_sender.py b/test/unit/test_face_detection_sender.py new file mode 100644 index 0000000..41ebd6c --- /dev/null +++ b/test/unit/test_face_detection_sender.py @@ -0,0 +1,170 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +import json +import mock +import pytest + +from robot_interface.endpoints.face_detector import FaceDetectionSender +from robot_interface.state import state + + +@pytest.fixture(autouse=True) +def initialized_state(monkeypatch): + """ + Fully initialize global state so __getattribute__ allows access. + """ + # Bypass the initialization guard + monkeypatch.setattr(state, "is_initialized", True, raising=False) + + # Install a controllable exit_event + exit_event = mock.Mock() + exit_event.is_set = mock.Mock(return_value=True) + monkeypatch.setattr(state, "exit_event", exit_event, raising=False) + + # Default qi_session is None unless overridden + monkeypatch.setattr(state, "qi_session", None, raising=False) + + yield + + +def test_start_face_detection_no_qi_session(): + """ + Returns early when qi_session is None. + """ + sender = FaceDetectionSender(mock.Mock()) + sender.start_face_detection() + + assert sender._face_thread is None + assert sender._face_service is None + assert sender._memory_service is None + + +def test_start_face_detection_happy_path(mocker): + """ + Initializes services and starts background thread. + """ + mock_face = mock.Mock() + mock_memory = mock.Mock() + + mock_qi = mock.Mock() + mock_qi.service.side_effect = lambda name: { + "ALFaceDetection": mock_face, + "ALMemory": mock_memory, + }[name] + + state.qi_session = mock_qi + + fake_thread = mock.Mock() + mocker.patch("threading.Thread", return_value=fake_thread) + + sender = FaceDetectionSender(mock.Mock()) + sender.start_face_detection() + + mock_face.setTrackingEnabled.assert_called_with(False) + mock_face.setRecognitionEnabled.assert_called_with(False) + mock_face.subscribe.assert_called_once() + fake_thread.start.assert_called_once() + + +def test_face_loop_face_detected_true(mocker): + """ + Sends face_detected=True when face data exists. + """ + sender = FaceDetectionSender(mock.Mock()) + + sender._memory_service = mock.Mock() + sender._memory_service.getData.return_value = [0, [[1]]] + sender.socket = mock.Mock() + + mocker.patch("time.sleep") + state.exit_event.is_set.side_effect = [False, True] + + sender._face_loop() + + sent = sender.socket.send.call_args[0][0] + payload = json.loads(sent.decode("utf-8")) + + assert payload["face_detected"] is True + + +def test_face_loop_face_detected_false(mocker): + """ + Sends face_detected=False when no face data exists. + """ + sender = FaceDetectionSender(mock.Mock()) + + sender._memory_service = mock.Mock() + sender._memory_service.getData.return_value = [] + sender.socket = mock.Mock() + + mocker.patch("time.sleep") + state.exit_event.is_set.side_effect = [False, True] + + sender._face_loop() + + sent = sender.socket.send.call_args[0][0] + payload = json.loads(sent.decode("utf-8")) + + assert not payload["face_detected"] + + +def test_face_loop_handles_exception(mocker): + """ + Exceptions inside loop are swallowed. + """ + sender = FaceDetectionSender(mock.Mock()) + + sender._memory_service = mock.Mock() + sender._memory_service.getData.side_effect = Exception("boom") + sender.socket = mock.Mock() + + mocker.patch("time.sleep") + state.exit_event.is_set.side_effect = [False, True] + + # Must not raise + sender._face_loop() + + +def test_stop_face_detection_happy_path(): + """ + Unsubscribes and disables tracking. + """ + sender = FaceDetectionSender(mock.Mock()) + + mock_face = mock.Mock() + sender._face_service = mock_face + + sender.stop_face_detection() + + mock_face.unsubscribe.assert_called_once() + mock_face.setTrackingEnabled.assert_called_with(False) + + +def test_stop_face_detection_exception(): + """ + stop_face_detection swallows service exceptions. + """ + sender = FaceDetectionSender(mock.Mock()) + + mock_face = mock.Mock() + mock_face.unsubscribe.side_effect = Exception("fail") + sender._face_service = mock_face + + sender.stop_face_detection() + + +def test_close_calls_stop_face_detection(mocker): + """ + close() calls parent close and stop_face_detection(). + """ + sender = FaceDetectionSender(mock.Mock()) + + mocker.patch.object(sender, "stop_face_detection") + mocker.patch( + "robot_interface.endpoints.face_detector.SocketBase.close" + ) + + sender.close() + + sender.stop_face_detection.assert_called_once() diff --git a/test/unit/test_main.py b/test/unit/test_main.py index f323630..afd4b05 100644 --- a/test/unit/test_main.py +++ b/test/unit/test_main.py @@ -55,6 +55,9 @@ class DummySender: def start(self): self.called = True + def start_face_detection(self): + self.called = True + def close(self): pass @@ -108,11 +111,13 @@ def patched_main_components(monkeypatch, fake_sockets, fake_poll): fake_act = FakeReceiver(act_sock) video_sender = DummySender() audio_sender = DummySender() + face_sender = DummySender() monkeypatch.setattr(main_mod, "MainReceiver", lambda ctx: fake_main) monkeypatch.setattr(main_mod, "ActuationReceiver", lambda ctx: fake_act) monkeypatch.setattr(main_mod, "VideoSender", lambda ctx: video_sender) monkeypatch.setattr(main_mod, "AudioSender", lambda ctx: audio_sender) + monkeypatch.setattr(main_mod, "FaceDetectionSender", lambda ctx: face_sender) # Register sockets for the fake poller fake_poll.registered = {main_sock: zmq.POLLIN, act_sock: zmq.POLLIN}