chore: implemented test video function

If there is no qi session, the webcam of the device is used to send video.

requirements.txt

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

src/robot_interface/endpoints/actuation_receiver.py

@@ -1,6 +1,5 @@
 from __future__ import unicode_literals  # So that we can log texts with Unicode characters
 import logging
-import time
 from threading import Thread
 
 import Queue
@@ -8,7 +7,6 @@ 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
 
@@ -36,11 +34,8 @@ 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):
         """
@@ -67,7 +62,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)
@@ -75,7 +70,6 @@ class ActuationReceiver(ReceiverBase):
         else: 
             self._message_queue.put(text)
 
-
     def clear_queue(self):
         """
         Safely drains all pending messages from the queue.
@@ -89,21 +83,6 @@ 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.
@@ -144,16 +123,12 @@ 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("Adding single gesture to queue: {}".format(gesture))
+            logging.debug("Playing single gesture: {}".format(gesture))
+            getattr(qi, "async")(self._animation_service.run, gesture)
         else:
-            logging.debug("Adding tag gesture to queue: {}".format(gesture))
-        self._gesture_queue.put(gesture)
-
+            logging.debug("Playing tag gesture: {}".format(gesture))
+            getattr(qi, "async")(self._animation_service.runTag, gesture)
 
     def handle_message(self, message):
         """
@@ -173,26 +148,14 @@ 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.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()       
+                logging.error("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):
         """

src/robot_interface/endpoints/audio_sender.py

@@ -89,6 +89,7 @@ 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)

src/robot_interface/endpoints/video_sender.py

@@ -1,11 +1,13 @@
 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.
@@ -27,7 +29,9 @@ class VideoSender(SocketBase):
         Will not start if no qi session is available.
         """
         if not state.qi_session:
-            logging.info("No Qi session available. Not starting video loop.")
+            logging.info("No Qi session available. Starting video from webcam.")
+            thread = threading.Thread(target=self.test_video_stream)
+            thread.start()
             return
 
         video = state.qi_session.service("ALVideoDevice")
@@ -58,3 +62,29 @@ 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()

test/unit/test_actuation_receiver.py

@@ -1,13 +1,25 @@
 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
@@ -42,6 +54,7 @@ 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.
@@ -57,6 +70,7 @@ 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
@@ -214,7 +228,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, True , True, True]
+    mock_state.exit_event.is_set.side_effect = [False, True]
 
     # Put an item in the queue
     receiver._message_queue.put("Hello World")
@@ -229,24 +243,40 @@ def test_handle_messages_loop(mocker):
     assert mock_state.is_speaking is True
 
 
-def test_handle_gestures_runtime_error(mocker):
+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
     mocker.patch("threading.Thread")
+
+    # Mock the state object
     mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-    
-    # Use a side_effect that returns False then True thereafter
-    mock_state.exit_event.is_set.side_effect = [False, True, True, True, True]
+        
+    # 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
     
     mock_zmq_ctx = mock.Mock()
     receiver = ActuationReceiver(mock_zmq_ctx)
     
-    # ... rest of your setup ...
-    mock_anim = mock.Mock()
-    mock_anim.run.side_effect = RuntimeError("Wifi Lost")
-    receiver._animation_service = mock_anim
+    # This ensures the while loop body runs exactly once for our test
+    mock_state.exit_event.is_set.side_effect = [False, True]
+
+    # 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)
     
-    receiver._gesture_queue.put("wave")
-    
-    receiver._handle_gestures()
+    # 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
+
 
 def test_handle_messages_runtime_error(mocker):
     """
@@ -266,7 +296,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, True, True ]
+    mock_state.exit_event.is_set.side_effect = [False, True]
 
     receiver._message_queue.put("Test Message")
     
@@ -304,61 +334,54 @@ def test_clear_queue(mocker):
     # Assert the queue is empty
     assert receiver._message_queue.qsize() == 0
 
-def test_gesture_no_data(mocker):
-    mock_zmq = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq)
+def test_gesture_no_data(zmq_context, mocker):
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": ""}, True)
     # Just ensuring no crash
 
 
-def test_gesture_invalid_data(mocker):
-    mock_zmq = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq)
+def test_gesture_invalid_data(zmq_context, mocker):
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": 123}, True)
     # No crash expected
 
 
-def test_gesture_single_not_found(mocker):
-    mock_zmq = mock.Mock()
+def test_gesture_single_not_found(zmq_context, mocker):
     mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
     mock_tags.single_gestures = ["wave", "bow"]  # allowed single gestures
-    receiver = ActuationReceiver(mock_zmq)
+
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "unknown_gesture"}, True)
     # No crash expected
 
 
-def test_gesture_tag_not_found(mocker):
-    mock_zmq = mock.Mock()
+def test_gesture_tag_not_found(zmq_context, mocker):
     mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
     mock_tags.tags = ["happy", "sad"]
 
-    receiver = ActuationReceiver(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "not_a_tag"}, False)
     # No crash expected
 
 
-def test_gesture_no_qi_session( mocker):
-    mock_zmq = mock.Mock()
+def test_gesture_no_qi_session(zmq_context, mocker):
     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(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "hello"}, True)
     # No crash, path returns early
 
 
-def test_gesture_single_success(mocker):
-    mock_zmq = mock.Mock()
+def test_gesture_single_success(zmq_context, mocker):
     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"]
 
@@ -366,25 +389,17 @@ def test_gesture_single_success(mocker):
     mock_state.qi_session = mock.Mock()
     mock_state.qi_session.service.return_value = mock_animation_service
 
-    receiver = ActuationReceiver(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/single", "data": "wave"}, True)
 
-    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)
+    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"
 
 
-def test_gesture_tag_success(mocker):
-    mock_zmq = mock.Mock()
+def test_gesture_tag_success(zmq_context, mocker):
     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
 
@@ -395,26 +410,21 @@ def test_gesture_tag_success(mocker):
     mock_state.qi_session = mock.Mock()
     mock_state.qi_session.service.return_value = mock_animation_service
 
-    receiver = ActuationReceiver(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
     receiver._handle_gesture({"endpoint": "actuate/gesture/tag", "data": "greeting"}, False)
 
-    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)
+    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"
 
 
-def test_handle_message_all_routes(mocker):
+def test_handle_message_all_routes(zmq_context, mocker):
     """
     Ensures all handle_message endpoint branches route correctly.
     """
-    mock_zmq = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
+
     mock_speech = mocker.patch.object(receiver, "_handle_speech")
     mock_gesture = mocker.patch.object(receiver, "_handle_gesture")
 
@@ -426,13 +436,12 @@ def test_handle_message_all_routes(mocker):
     assert mock_gesture.call_count == 2
 
 
-def test_endpoint_description(mocker):
-    mock_zmq = mock.Mock()
+def test_endpoint_description(zmq_context, mocker):
     mock_tags = mocker.patch("robot_interface.endpoints.actuation_receiver.GestureTags")
     mock_tags.tags = ["happy"]
     mock_tags.single_gestures = ["wave"]
 
-    receiver = ActuationReceiver(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
     desc = receiver.endpoint_description()
 
     assert "gestures" in desc
@@ -442,20 +451,26 @@ def test_endpoint_description(mocker):
     assert desc["single_gestures"] == ["wave"]
 
 
-def test_gesture_single_real_gesturetags(mocker):
-    mock_zmq = mock.Mock()
+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
     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(mock_zmq)
+    receiver = ActuationReceiver(zmq_context)
 
+    # 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]
 
@@ -464,85 +479,8 @@ def test_gesture_single_real_gesturetags(mocker):
         is_single=True,
     )
 
-    time.sleep(0.2)
-    mock_state.qi_session.service.assert_called_with("ALAnimationPlayer")
-    mock_animation_service.run.assert_called_with(gesture)
+    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
 
-    # 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()

test/unit/test_audio_sender.py

@@ -77,7 +77,8 @@ 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
@@ -93,6 +94,36 @@ 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.

test/unit/test_video_sender.py

@@ -97,3 +97,120 @@ 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()