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

test: fixed video_sender tests

Browse Source 
ref: N25B-393
This commit is contained in:
Storm
2026-01-29 12:16:48 +01:00
parent 4a2cace1cf
commit 56becd84ac
2 changed files with 14 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/robot_interface/endpoints/video_sender.py
View File

@@ -3,6 +3,7 @@ import threading
import logging
import numpy as np
import struct
from robot_interface.endpoints.socket_base import SocketBase
from robot_interface.state import state
@@ -63,8 +64,8 @@ class VideoSender(SocketBase):
width = img[0]
height = img[1]
width_bytes = width.to_bytes(4, 'little')
height_bytes = height.to_bytes(4, 'little')
width_bytes = struct.pack('<I', width)
height_bytes = struct.pack('<I', height)
self.socket.send_multipart([width_bytes, height_bytes, raw_data])
except:

15
test/unit/test_video_sender.py
View File

@@ -1,5 +1,6 @@
# coding=utf-8
import struct
import mock
import pytest
import zmq
@@ -48,8 +49,10 @@ def test_video_streaming(zmq_context, mocker):
# Pepper's image buffer lives at index 6
mocker.patch.object(settings.video_config, "image_buffer", 6)
test_width = 320
test_height = 240
mock_video_service = mock.Mock()
mock_video_service.getImageRemote.return_value = [None]*6 + ["fake_img"]
mock_video_service.getImageRemote.return_value = [test_width, test_height, None, None, None, None, b"fake_img"]
fake_session = mock.Mock()
fake_session.service.return_value = mock_video_service
@@ -63,12 +66,16 @@ def test_video_streaming(zmq_context, mocker):
sender = VideoSender(zmq_context)
send_socket = mock.Mock()
sender.socket.send = send_socket
sender.socket.send_multipart = send_socket
sender.start_video_rcv()
sender.video_rcv_loop(mock_video_service, "stream_name")
send_socket.assert_called_with("fake_img")
send_socket.assert_called_with([
struct.pack('<I', 320),
struct.pack('<I', 240),
b"fake_img"
])
def test_video_receive_error(zmq_context, mocker):
@@ -91,7 +98,7 @@ def test_video_receive_error(zmq_context, mocker):
sender = VideoSender(zmq_context)
send_socket = mock.Mock()
sender.socket.send = send_socket
sender.socket.send_multipart = send_socket
sender.start_video_rcv()
sender.video_rcv_loop(mock_video_service, "stream_name")