112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
import zmq
|
|
from aarecommon.models.models import DAQStatusModel
|
|
from PySide6.QtGui import QPixmap
|
|
|
|
from aare.gui.threads.camera_thread import SampleCameraThread
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_zmq():
|
|
with patch("zmq.Context") as mock_ctx_class:
|
|
mock_ctx = mock_ctx_class.return_value
|
|
mock_socket = mock_ctx.socket.return_value
|
|
yield mock_socket
|
|
|
|
|
|
@pytest.fixture
|
|
def camera_thread(mock_zmq, qtbot):
|
|
thread = SampleCameraThread("tcp://127.0.0.1:5555")
|
|
yield thread
|
|
thread.running = False
|
|
try:
|
|
thread.stop()
|
|
except RuntimeError:
|
|
pass
|
|
|
|
|
|
def test_init(camera_thread):
|
|
assert camera_thread.running is True
|
|
assert camera_thread._SampleCameraThread__camera_available is False
|
|
|
|
|
|
def test_update_daq_status(camera_thread):
|
|
s = MagicMock()
|
|
s.geom.beam_location_pxl.x = 100
|
|
s.geom.beam_location_pxl.y = 200
|
|
camera_thread.update_daq_status(s)
|
|
assert camera_thread._SampleCameraThread__beam_x == 100
|
|
assert camera_thread._SampleCameraThread__beam_y == 200
|
|
|
|
|
|
def test_enable_focus_measurement(camera_thread):
|
|
camera_thread.enable_focus_measurement(True)
|
|
assert camera_thread._SampleCameraThread__measure_focus is True
|
|
camera_thread.enable_focus_measurement(False)
|
|
assert camera_thread._SampleCameraThread__measure_focus is False
|
|
|
|
|
|
def test_run_success(camera_thread, mock_zmq, qtbot):
|
|
img = np.zeros((10, 10, 3), dtype=np.uint8)
|
|
_, jpeg_bytes = cv2.imencode(".jpg", img)
|
|
header = json.dumps({"encoding": "jpeg"}).encode("utf-8")
|
|
|
|
def side_effect():
|
|
camera_thread.running = False
|
|
return [header, jpeg_bytes.tobytes()]
|
|
|
|
mock_zmq.recv_multipart.side_effect = side_effect
|
|
|
|
with qtbot.waitSignal(camera_thread.camera_image, timeout=5000):
|
|
camera_thread.run()
|
|
|
|
assert camera_thread._SampleCameraThread__camera_available is True
|
|
|
|
|
|
def test_run_bayer_success(camera_thread, mock_zmq, qtbot):
|
|
h, w = 10, 10
|
|
raw = np.zeros((h, w), dtype=np.uint8)
|
|
header = json.dumps({"shape": [h, w]}).encode("utf-8")
|
|
|
|
def side_effect():
|
|
camera_thread.running = False
|
|
return [header, raw.tobytes()]
|
|
|
|
mock_zmq.recv_multipart.side_effect = side_effect
|
|
|
|
with qtbot.waitSignal(camera_thread.camera_image, timeout=5000):
|
|
camera_thread.run()
|
|
|
|
assert camera_thread._SampleCameraThread__camera_available is True
|
|
|
|
|
|
def test_run_zmq_timeout(camera_thread, mock_zmq, qtbot):
|
|
def side_effect():
|
|
camera_thread.running = False
|
|
raise zmq.Again()
|
|
|
|
mock_zmq.recv_multipart.side_effect = side_effect
|
|
camera_thread._SampleCameraThread__fps_emit_period_s = 0.0
|
|
|
|
with qtbot.waitSignal(camera_thread.fps_measure, timeout=5000):
|
|
try:
|
|
camera_thread.run()
|
|
except StopIteration:
|
|
pass
|
|
|
|
|
|
def test_stop(camera_thread, mock_zmq):
|
|
camera_thread.quit = MagicMock()
|
|
camera_thread.wait = MagicMock()
|
|
|
|
camera_thread.stop()
|
|
|
|
assert camera_thread.running is False
|
|
if hasattr(camera_thread, "_SampleCameraThread__socket"):
|
|
mock_zmq.close.assert_called()
|