7 Commits

Author SHA1 Message Date
Pim Hutting
eab2481b85 chore: fixed flakiness of tests 2026-01-14 16:34:34 +01:00
Pim Hutting
a55acd57b6 chore: finalized tests, added queue
ref: N25B-387
2025-12-18 12:55:09 +01:00
Storm
5d5c8553c2 test: added test configuration to always mock zmq context to fix zmq port congestion issues
ref: N25B-386
2025-12-16 16:37:36 +01:00
Storm
79db2c77c8 feat: added queue-size log message
ref: N25B-386
2025-12-16 16:17:09 +01:00
Storm
b6f2893c25 style: moved one line
ref: N25B-386
2025-12-16 14:54:02 +01:00
Storm
b3e3a1eb80 feat: implemented force speech functionality in RI and refactored actuation_receiver tests
Before actuation_receiver tests started a receiver with real zmq context. This led to flaky tests because of port congestion issues.

ref: N25B-386
2025-12-16 14:49:44 +01:00
Storm
912af8d821 feat: implemented force speech functionality in RI and refactored actuation_receiver tests
Before actuation_receiver tests started a receiver with real zmq context. This led to flaky tests because of port congestion issues.

ref: N25B-386
2025-12-12 14:38:06 +01:00
7 changed files with 198 additions and 279 deletions

View File

@@ -7,4 +7,3 @@ sphinx
sphinx_rtd_theme
pre-commit
python-dotenv
opencv-python==4.1.2.30

View File

@@ -1,5 +1,6 @@
from __future__ import unicode_literals # So that we can log texts with Unicode characters
import logging
import time
from threading import Thread
import Queue
@@ -7,6 +8,7 @@ import zmq
from robot_interface.endpoints.receiver_base import ReceiverBase
from robot_interface.state import state
from robot_interface.core.config import settings
from robot_interface.endpoints.gesture_settings import GestureTags
@@ -34,8 +36,11 @@ class ActuationReceiver(ReceiverBase):
self._tts_service = None
self._animation_service = None
self._message_queue = Queue.Queue()
self._gesture_queue = Queue.Queue()
self.message_thread = Thread(target=self._handle_messages)
self.message_thread.start()
self.gesture_thread = Thread(target=self._handle_gestures)
self.gesture_thread.start()
def _handle_speech(self, message):
"""
@@ -62,7 +67,7 @@ class ActuationReceiver(ReceiverBase):
if not self._tts_service:
self._tts_service = state.qi_session.service("ALTextToSpeech")
if message.get("is_priority"):
if (message.get("is_priority")):
# Bypass queue and speak immediately
self.clear_queue()
self._message_queue.put(text)
@@ -70,6 +75,7 @@ class ActuationReceiver(ReceiverBase):
else:
self._message_queue.put(text)
def clear_queue(self):
"""
Safely drains all pending messages from the queue.
@@ -83,6 +89,21 @@ class ActuationReceiver(ReceiverBase):
pass
logging.info("Message queue cleared.")
def clear_gesture_queue(self):
"""
Safely drains all pending gestures from the gesture queue.
"""
logging.info("Gesture queue size: {}".format(self._gesture_queue.qsize()))
try:
while True:
# Remove items one by one without waiting
self._gesture_queue.get_nowait()
except Queue.Empty:
pass
logging.info("Gesture queue cleared.")
logging.info("Gesture queue size: {}".format(self._gesture_queue.qsize()))
def _handle_gesture(self, message, is_single):
"""
Handle a gesture actuation request.
@@ -123,12 +144,16 @@ class ActuationReceiver(ReceiverBase):
# Play the gesture. Pepper comes with predefined animations like "Wave", "Greet", "Clap"
# You can also create custom animations using Choregraphe and upload them to the robot.
if (message.get("is_priority")):
# Clear queue and play
self.clear_gesture_queue()
logging.debug("Force playing gesture immediately: {}".format(gesture))
if is_single:
logging.debug("Playing single gesture: {}".format(gesture))
getattr(qi, "async")(self._animation_service.run, gesture)
logging.debug("Adding single gesture to queue: {}".format(gesture))
else:
logging.debug("Playing tag gesture: {}".format(gesture))
getattr(qi, "async")(self._animation_service.runTag, gesture)
logging.debug("Adding tag gesture to queue: {}".format(gesture))
self._gesture_queue.put(gesture)
def handle_message(self, message):
"""
@@ -148,13 +173,25 @@ class ActuationReceiver(ReceiverBase):
while not state.exit_event.is_set():
try:
text = self._message_queue.get(timeout=0.1)
if not state.is_speaking: print("Started speaking.")
state.is_speaking = True
self._tts_service.say(text)
except Queue.Empty:
if state.is_speaking: print("Finished speaking.")
state.is_speaking = False
except RuntimeError:
logging.error("Lost connection to Pepper. Please check if you're connected to the "
"local WiFi and restart this application.")
logging.warn("Lost connection to Pepper. Please check if you're connected to the local WiFi and restart this application.")
state.exit_event.set()
def _handle_gestures(self):
while not state.exit_event.is_set():
try:
gesture = self._gesture_queue.get(timeout=0.1)
self._animation_service.run(gesture)
except Queue.Empty:
pass
except RuntimeError:
logging.warn("Lost connection to Pepper. Please check if you're connected to the local WiFi and restart this application.")
state.exit_event.set()
def endpoint_description(self):

View File

@@ -89,7 +89,6 @@ class AudioSender(SocketBase):
try:
while not state.exit_event.is_set():
data = stream.read(chunk)
if (state.is_speaking): continue # Do not send audio while the robot is speaking
self.socket.send(data)
except IOError as e:
logger.error("Stopped listening: failed to get audio from microphone.", exc_info=e)

View File

@@ -1,13 +1,11 @@
import zmq
import threading
import logging
import cv2
from robot_interface.endpoints.socket_base import SocketBase
from robot_interface.state import state
from robot_interface.core.config import settings
class VideoSender(SocketBase):
"""
Video sender endpoint, responsible for sending video frames.
@@ -29,9 +27,7 @@ class VideoSender(SocketBase):
Will not start if no qi session is available.
"""
if not state.qi_session:
logging.info("No Qi session available. Starting video from webcam.")
thread = threading.Thread(target=self.test_video_stream)
thread.start()
logging.info("No Qi session available. Not starting video loop.")
return
video = state.qi_session.service("ALVideoDevice")
@@ -62,29 +58,3 @@ class VideoSender(SocketBase):
self.socket.send(img[settings.video_config.image_buffer])
except:
logging.warn("Failed to retrieve video image from robot.")
def test_video_stream(self):
"""
Test function to send video from local webcam instead of the robot.
"""
cap = cv2.VideoCapture(0)
if not cap.isOpened():
logging.error("Could not open webcam for video stream test.")
return
while not state.exit_event.is_set():
ret, frame = cap.read()
if not ret:
logging.warning("Failed to read frame from webcam.")
continue
cv2.waitKey(1)
small_frame = cv2.resize(frame, (320, 240), interpolation=cv2.INTER_AREA)
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 70]
_, buffer = cv2.imencode('.jpg', small_frame, encode_param)
self.socket.send(buffer.tobytes())
cap.release()

View File

@@ -1,25 +1,13 @@
import sys
import time
import mock
import pytest
import zmq
import Queue
from robot_interface.endpoints.actuation_receiver import ActuationReceiver
from robot_interface.endpoints.gesture_settings import GestureTags
@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_force_speech_clears_queue(mocker):
"""
Tests that a force speech message clears the existing queue
@@ -54,7 +42,6 @@ def test_force_speech_clears_queue(mocker):
queued_item = receiver._message_queue.get()
assert queued_item == "Emergency Notification"
def test_handle_unimplemented_endpoint(mocker):
"""
Tests handling of unknown endpoints.
@@ -70,7 +57,6 @@ def test_handle_unimplemented_endpoint(mocker):
"data": None,
})
def test_speech_message_no_data(mocker):
"""
Tests that if the message data is empty, the receiver returns immediately
@@ -228,7 +214,7 @@ def test_handle_messages_loop(mocker):
receiver._tts_service = mock_tts_service
# This ensures the while loop iterates exactly once
mock_state.exit_event.is_set.side_effect = [False, True]
mock_state.exit_event.is_set.side_effect = [False, True, True , True, True]
# Put an item in the queue
receiver._message_queue.put("Hello World")
@@ -243,40 +229,24 @@ def test_handle_messages_loop(mocker):
assert mock_state.is_speaking is True
def test_handle_messages_queue_empty(mocker):
"""
Tests the Queue.Empty exception handler in the consumer loop.
This covers the logic that resets 'state.is_speaking' to False.
"""
# Prevent the real background thread from starting
def test_handle_gestures_runtime_error(mocker):
mocker.patch("threading.Thread")
# Mock the state object
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
# Setup 'is_speaking' property mock
# We set return_value=True so the code enters the 'if state.is_speaking:' block.
# We use PropertyMock to track when this attribute is set.
type(mock_state).is_speaking = True
# Use a side_effect that returns False then True thereafter
mock_state.exit_event.is_set.side_effect = [False, True, True, True, True]
mock_zmq_ctx = mock.Mock()
receiver = ActuationReceiver(mock_zmq_ctx)
# This ensures the while loop body runs exactly once for our test
mock_state.exit_event.is_set.side_effect = [False, True]
# ... rest of your setup ...
mock_anim = mock.Mock()
mock_anim.run.side_effect = RuntimeError("Wifi Lost")
receiver._animation_service = mock_anim
# Force get() to raise Queue.Empty immediately (simulate timeout)
# We patch the 'get' method on the specific queue instance of our receiver
#mocker.patch.object(receiver._message_queue, 'get', side_effect=Queue.Empty)
# Run the loop logic manually (synchronously)
receiver._handle_messages()
# Final Assertion: Verify is_speaking was set to False
# The code execution order is: read (returns True) -> print -> set (to False)
# assert_called_with checks the arguments of the LAST call, which is the setter.
assert mock_state.is_speaking is False
receiver._gesture_queue.put("wave")
receiver._handle_gestures()
def test_handle_messages_runtime_error(mocker):
"""
@@ -296,7 +266,7 @@ def test_handle_messages_runtime_error(mocker):
# Initialize receiver with the mock context
receiver = ActuationReceiver(mock_zmq_ctx)
mock_state.exit_event.is_set.side_effect = [False, True]
mock_state.exit_event.is_set.side_effect = [False, True, True, True ]
receiver._message_queue.put("Test Message")
@@ -334,54 +304,61 @@ def test_clear_queue(mocker):
# Assert the queue is empty
assert receiver._message_queue.qsize() == 0
def test_gesture_no_data(zmq_context, mocker):
receiver = ActuationReceiver(zmq_context)
def test_gesture_no_data(mocker):
mock_zmq = mock.Mock()
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": ""}, True)
# Just ensuring no crash
def test_gesture_invalid_data(zmq_context, mocker):
receiver = ActuationReceiver(zmq_context)
def test_gesture_invalid_data(mocker):
mock_zmq = mock.Mock()
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": 123}, True)
# No crash expected
def test_gesture_single_not_found(zmq_context, mocker):
def test_gesture_single_not_found(mocker):
mock_zmq = mock.Mock()
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.single_gestures = ["wave", "bow"] # allowed single gestures
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "unknown_gesture"}, True)
# No crash expected
def test_gesture_tag_not_found(zmq_context, mocker):
def test_gesture_tag_not_found(mocker):
mock_zmq = mock.Mock()
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.tags = ["happy", "sad"]
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "not_a_tag"}, False)
# No crash expected
def test_gesture_no_qi_session(zmq_context, mocker):
def test_gesture_no_qi_session( mocker):
mock_zmq = mock.Mock()
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_state.qi_session = None
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.single_gestures = ["hello"]
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "hello"}, True)
# No crash, path returns early
def test_gesture_single_success(zmq_context, mocker):
def test_gesture_single_success(mocker):
mock_zmq = mock.Mock()
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
# Allow loops to run
mock_state.exit_event.is_set.return_value = False
mock_qi = mock.Mock()
sys.modules["qi"] = mock_qi
# Setup gesture settings
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.single_gestures = ["wave"]
@@ -389,17 +366,25 @@ def test_gesture_single_success(zmq_context, mocker):
mock_state.qi_session = mock.Mock()
mock_state.qi_session.service.return_value = mock_animation_service
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "wave"}, True)
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
getattr(mock_qi, "async").assert_called_once()
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.run
assert getattr(mock_qi, "async").call_args[0][1] == "wave"
time.sleep(0.2)
mock_state.qi_session.service.assert_called_with("ALAnimationPlayer")
mock_animation_service.run.assert_called_with("wave")
# CLEANUP: Signal exit AND join threads
mock_state.exit_event.is_set.return_value = True
receiver.message_thread.join(timeout=1.0)
receiver.gesture_thread.join(timeout=1.0)
def test_gesture_tag_success(zmq_context, mocker):
def test_gesture_tag_success(mocker):
mock_zmq = mock.Mock()
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_state.exit_event.is_set.return_value = False
mock_qi = mock.Mock()
sys.modules["qi"] = mock_qi
@@ -410,21 +395,26 @@ def test_gesture_tag_success(zmq_context, mocker):
mock_state.qi_session = mock.Mock()
mock_state.qi_session.service.return_value = mock_animation_service
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "greeting"}, False)
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
getattr(mock_qi, "async").assert_called_once()
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.runTag
assert getattr(mock_qi, "async").call_args[0][1] == "greeting"
time.sleep(0.2)
mock_state.qi_session.service.assert_called_with("ALAnimationPlayer")
mock_animation_service.run.assert_called_with("greeting")
# CLEANUP: Signal exit AND join threads
mock_state.exit_event.is_set.return_value = True
receiver.message_thread.join(timeout=1.0)
receiver.gesture_thread.join(timeout=1.0)
def test_handle_message_all_routes(zmq_context, mocker):
def test_handle_message_all_routes(mocker):
"""
Ensures all handle_message endpoint branches route correctly.
"""
receiver = ActuationReceiver(zmq_context)
mock_zmq = mock.Mock()
receiver = ActuationReceiver(mock_zmq)
mock_speech = mocker.patch.object(receiver, "_handle_speech")
mock_gesture = mocker.patch.object(receiver, "_handle_gesture")
@@ -436,12 +426,13 @@ def test_handle_message_all_routes(zmq_context, mocker):
assert mock_gesture.call_count == 2
def test_endpoint_description(zmq_context, mocker):
def test_endpoint_description(mocker):
mock_zmq = mock.Mock()
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.tags = ["happy"]
mock_tags.single_gestures = ["wave"]
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
desc = receiver.endpoint_description()
assert "gestures" in desc
@@ -451,26 +442,20 @@ def test_endpoint_description(zmq_context, mocker):
assert desc["single_gestures"] == ["wave"]
def test_gesture_single_real_gesturetags(zmq_context, mocker):
"""
Uses the real GestureTags (no mocking) to ensure the receiver
references GestureTags.single_gestures correctly.
"""
# Ensure qi session exists so we pass the early return
def test_gesture_single_real_gesturetags(mocker):
mock_zmq = mock.Mock()
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_state.qi_session = mock.Mock()
mock_state.exit_event.is_set.return_value = False
# Mock qi.async to avoid real async calls
mock_qi = mock.Mock()
sys.modules["qi"] = mock_qi
# Mock animation service
mock_animation_service = mock.Mock()
mock_state.qi_session.service.return_value = mock_animation_service
receiver = ActuationReceiver(zmq_context)
receiver = ActuationReceiver(mock_zmq)
# Pick a real gesture from GestureTags.single_gestures
assert len(GestureTags.single_gestures) > 0, "GestureTags.single_gestures must not be empty"
gesture = GestureTags.single_gestures[0]
@@ -479,8 +464,85 @@ def test_gesture_single_real_gesturetags(zmq_context, mocker):
is_single=True,
)
mock_state.qi_session.service.assert_called_once_with("ALAnimationPlayer")
getattr(mock_qi, "async").assert_called_once()
assert getattr(mock_qi, "async").call_args[0][0] == mock_animation_service.run
assert getattr(mock_qi, "async").call_args[0][1] == gesture
time.sleep(0.2)
mock_state.qi_session.service.assert_called_with("ALAnimationPlayer")
mock_animation_service.run.assert_called_with(gesture)
# CLEANUP: Signal exit AND join threads
mock_state.exit_event.is_set.return_value = True
receiver.message_thread.join(timeout=1.0)
receiver.gesture_thread.join(timeout=1.0)
def test_clear_gesture_queue(mocker):
# Prevent background threads from eating the items
mocker.patch("threading.Thread")
mock_zmq_ctx = mock.Mock()
receiver = ActuationReceiver(mock_zmq_ctx)
# Populate the queue
receiver._gesture_queue.put("gesture1")
receiver._gesture_queue.put("gesture2")
assert receiver._gesture_queue.qsize() == 2
# Clear the queue
receiver.clear_gesture_queue()
# Assert the queue is empty
assert receiver._gesture_queue.qsize() == 0
def test_gesture_priority_clears_queue(mocker):
mocker.patch("threading.Thread")
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
# Mock QI and Tags so valid checks pass
mock_qi = mock.Mock()
sys.modules["qi"] = mock_qi
mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
mock_tags.single_gestures = ["urgent_wave"]
# Setup Animation Service
mock_anim = mock.Mock()
mock_state.qi_session.service.return_value = mock_anim
mock_zmq_ctx = mock.Mock()
receiver = ActuationReceiver(mock_zmq_ctx)
# Pre-fill queue with "slow" gestures
receiver._gesture_queue.put("slow_gesture_1")
receiver._gesture_queue.put("slow_gesture_2")
assert receiver._gesture_queue.qsize() == 2
# Send priority gesture
priority_msg = {
"endpoint": "actuate/gesture/single",
"data": "urgent_wave",
"is_priority": True,
}
receiver._handle_gesture(priority_msg, is_single=True)
# Assert old items are gone and only new one remains
assert receiver._gesture_queue.qsize() == 1
assert receiver._gesture_queue.get() == "urgent_wave"
def test_handle_gestures_loop_empty(mocker):
mocker.patch("threading.Thread")
mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
mock_zmq_ctx = mock.Mock()
receiver = ActuationReceiver(mock_zmq_ctx)
# Run loop exactly once
mock_state.exit_event.is_set.side_effect = [False, True, True, True]
# We don't put anything in the queue, so .get(timeout=0.1) will raise Queue.Empty.
# The code should catch it and pass.
receiver._handle_gestures()
# If we reached here without raising an exception, the test passes.
# We can assert that the queue is still valid/empty.
assert receiver._gesture_queue.empty()

View File

@@ -78,7 +78,6 @@ def test_sending_audio(mocker):
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
@@ -94,36 +93,6 @@ def test_sending_audio(mocker):
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.

View File

@@ -97,120 +97,3 @@ def test_video_receive_error(zmq_context, mocker):
sender.video_rcv_loop(mock_video_service, "stream_name")
send_socket.assert_not_called()
def test_video_stream_camera_fail(zmq_context, mocker):
"""
Test that the function logs an error and returns early if
the webcam cannot be opened.
"""
_patch_basics(mocker)
# Mock cv2 and logging
mock_cv2 = mocker.patch("robot_interface.endpoints.video_sender.cv2")
mock_logging = mocker.patch("robot_interface.endpoints.video_sender.logging")
# Setup the mock capture to fail isOpened()
mock_cap = mock.Mock()
mock_cap.isOpened.return_value = False
mock_cv2.VideoCapture.return_value = mock_cap
sender = VideoSender(zmq_context)
sender.test_video_stream()
# Assertions
mock_cv2.VideoCapture.assert_called_with(0)
# Ensure the loop was never entered and cleanup didn't happen
assert not mock_cap.read.called
assert not mock_cap.release.called
def test_video_stream_read_fail(zmq_context, mocker):
"""
Test that the function logs a warning and continues the loop
if a specific frame fails to read.
"""
_patch_basics(mocker)
_patch_exit_event(mocker) # Run loop exactly once
mock_cv2 = mocker.patch("robot_interface.endpoints.video_sender.cv2")
mock_logging = mocker.patch("robot_interface.endpoints.video_sender.logging")
# Setup capture to open successfully, but fail the read()
mock_cap = mock.Mock()
mock_cap.isOpened.return_value = True
# Return (False, None) simulating a failed frame read
mock_cap.read.return_value = (False, None)
mock_cv2.VideoCapture.return_value = mock_cap
sender = VideoSender(zmq_context)
# Mock the socket to ensure nothing is sent
sender.socket = mock.Mock()
sender.test_video_stream()
# Ensure we skipped the processing steps
assert not mock_cv2.resize.called
assert not sender.socket.send.called
# Ensure cleanup happened at the end
mock_cap.release.assert_called_once()
def test_video_stream_success(zmq_context, mocker):
"""
Test the happy path: Frame read -> Resize -> Encode -> Send.
"""
_patch_basics(mocker)
_patch_exit_event(mocker) # Run loop exactly once
mock_cv2 = mocker.patch("robot_interface.endpoints.video_sender.cv2")
# Setup constants usually found in cv2
mock_cv2.IMWRITE_JPEG_QUALITY = 1
mock_cv2.INTER_AREA = 2
# Setup capture to work perfectly
mock_cap = mock.Mock()
mock_cap.isOpened.return_value = True
fake_frame = "original_frame_data"
mock_cap.read.return_value = (True, fake_frame)
mock_cv2.VideoCapture.return_value = mock_cap
# Setup Resize and Encode
mock_cv2.resize.return_value = "small_frame_data"
# Mock buffer behavior
mock_buffer = mock.Mock()
mock_buffer.tobytes.return_value = b"encoded_bytes"
# imencode returns (retval, buffer)
mock_cv2.imencode.return_value = (True, mock_buffer)
sender = VideoSender(zmq_context)
sender.socket = mock.Mock()
sender.test_video_stream()
# Assertions
# 1. Check waitKey (the 1ms delay)
mock_cv2.waitKey.assert_called_with(1)
# 2. Check Resize logic
mock_cv2.resize.assert_called_with(
fake_frame,
(320, 240),
interpolation=mock_cv2.INTER_AREA
)
# 3. Check Encode logic
mock_cv2.imencode.assert_called_with(
'.jpg',
"small_frame_data",
[mock_cv2.IMWRITE_JPEG_QUALITY, 70]
)
# 4. Check Socket Send
sender.socket.send.assert_called_with(b"encoded_bytes")
# 5. Check Cleanup
mock_cap.release.assert_called_once()