84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
from PySide6.QtGui import QImage
|
|
|
|
from aare.gui.threads.axis_video_thread import VideoThread
|
|
|
|
|
|
@pytest.fixture
|
|
def video_thread(qtbot):
|
|
thread = VideoThread("127.0.0.1", camera=1)
|
|
return thread
|
|
|
|
|
|
def test_init(video_thread):
|
|
assert video_thread.camera_ip == "127.0.0.1"
|
|
assert video_thread.camera == 1
|
|
assert video_thread.running is False
|
|
assert video_thread.is_busy is False
|
|
|
|
|
|
def test_set_camera_ip(video_thread):
|
|
video_thread.set_camera_ip("192.168.1.1")
|
|
assert video_thread.camera_ip == "192.168.1.1"
|
|
|
|
|
|
def test_set_busy(video_thread):
|
|
video_thread.set_busy(True)
|
|
assert video_thread.is_busy is True
|
|
|
|
|
|
def test_process_buffer_success(video_thread, qtbot):
|
|
# Create a small valid JPEG
|
|
img = np.zeros((10, 10, 3), dtype=np.uint8)
|
|
_, jpeg_bytes = cv2.imencode(".jpg", img)
|
|
jpeg_bytes = jpeg_bytes.tobytes()
|
|
|
|
boundary = b"--boundary"
|
|
buffer = b"--boundary\r\nContent-Type: image/jpeg\r\n\r\n" + jpeg_bytes + b"\r\n--boundary"
|
|
|
|
with qtbot.waitSignal(video_thread.frame_ready, timeout=1000) as blocker:
|
|
video_thread._process_buffer(buffer, boundary)
|
|
|
|
assert isinstance(blocker.args[0], QImage)
|
|
assert blocker.args[0].width() == 10
|
|
assert blocker.args[0].height() == 10
|
|
|
|
|
|
def test_process_buffer_invalid_jpeg(video_thread):
|
|
boundary = b"--boundary"
|
|
buffer = b"--boundary\r\nContent-Type: image/jpeg\r\n\r\nNOT_A_JPEG\r\n--boundary"
|
|
|
|
# Should not emit anything or raise exception
|
|
video_thread._process_buffer(buffer, boundary)
|
|
|
|
|
|
def test_stop(video_thread):
|
|
mock_session = MagicMock()
|
|
video_thread.session = mock_session
|
|
# Mocking wait and quit to avoid actual thread blocking in test
|
|
video_thread.quit = MagicMock()
|
|
video_thread.wait = MagicMock()
|
|
|
|
video_thread.stop()
|
|
|
|
assert video_thread.running is False
|
|
assert video_thread.session is None
|
|
mock_session.close.assert_called_once()
|
|
video_thread.quit.assert_called_once()
|
|
video_thread.wait.assert_called_once_with(5000)
|
|
|
|
|
|
@patch("requests.Session")
|
|
def test_run_connection_error(mock_session_class, video_thread, qtbot):
|
|
mock_session = mock_session_class.return_value
|
|
mock_session.get.side_effect = Exception("Connection Refused")
|
|
|
|
with qtbot.waitSignal(video_thread.error_occurred, timeout=1000) as blocker:
|
|
video_thread.run()
|
|
|
|
assert "Unexpected error" in blocker.args[0]
|