636 lines
22 KiB
Python
636 lines
22 KiB
Python
import displayarray.window.subscriber_windows as sub_win
|
|
from threading import Thread
|
|
import mock
|
|
import cv2
|
|
import numpy as np
|
|
from displayarray.effects.select_channels import SelectChannels
|
|
import pytest
|
|
|
|
'''
|
|
OpenCV now has a new error:
|
|
cv2.error: OpenCV(4.5.3) C:\...\window_w32.cpp:2559: error: (-27:Null pointer) NULL window:
|
|
'displayarray (press ESC to quit)' in function 'cvSetMouseCallback'
|
|
The code works if the window's aren't mocked, but until the error is bypassed,
|
|
this can't be tested in continuous integration.
|
|
|
|
def test_init_defaults():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow") as window_mock:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
assert sw.source_names == ["0"]
|
|
assert sw.input_vid_global_names == ["0"]
|
|
assert sw.window_names == ["displayarray"]
|
|
assert sw.input_cams == [0]
|
|
assert sw.exited == False
|
|
|
|
window_mock.assert_called_once_with("displayarray (press ESC to quit)")
|
|
|
|
|
|
def test_init():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow") as window_mock:
|
|
cb = mock.MagicMock()
|
|
sw = sub_win.SubscriberWindows(["test name"], [1], cb)
|
|
|
|
assert sw.source_names == ["1"]
|
|
assert sw.input_vid_global_names == ["1"]
|
|
assert sw.window_names == ["test name"]
|
|
assert sw.input_cams == [1]
|
|
assert sw.exited == False
|
|
|
|
window_mock.assert_called_once_with("test name (press ESC to quit)")
|
|
|
|
|
|
def test_bool():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "update"
|
|
) as mock_update:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
mock_update.assert_called_once()
|
|
mock_update.reset_mock()
|
|
|
|
assert bool(sw) == True
|
|
|
|
mock_update.assert_called_once()
|
|
mock_update.reset_mock()
|
|
|
|
sw.exited = True
|
|
|
|
assert bool(sw) == False
|
|
|
|
|
|
def test_block():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "update"
|
|
), mock.patch.object(sub_win.SubscriberWindows, "loop") as mock_loop:
|
|
sw = sub_win.SubscriberWindows()
|
|
sw.close_threads.append(mock.MagicMock())
|
|
|
|
sw.block()
|
|
|
|
mock_loop.assert_called_once()
|
|
sw.close_threads[0].join.assert_called_once()
|
|
|
|
|
|
def test_add_source():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"):
|
|
sw = sub_win.SubscriberWindows().add_source(2)
|
|
|
|
assert sw.source_names == ["0", "2"]
|
|
assert sw.input_vid_global_names == ["0", "2"]
|
|
assert sw.input_cams == [0, 2]
|
|
|
|
|
|
def test_add_callback():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"):
|
|
mock_cb = mock.MagicMock()
|
|
|
|
sw = sub_win.SubscriberWindows().add_callback(mock_cb)
|
|
|
|
assert sw.callbacks[0] == mock_cb
|
|
|
|
|
|
def test_handle_keys():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch(
|
|
"displayarray.window.subscriber_windows.subscriber_dictionary.stop_cam"
|
|
) as mock_stop, mock.patch(
|
|
"displayarray.window.subscriber_windows.warnings"
|
|
) as mock_warnings, mock.patch(
|
|
"displayarray.window.subscriber_windows.RuntimeWarning"
|
|
) as mock_runtime, mock.patch.object(
|
|
cv2, "destroyWindow"
|
|
) as mock_destroy:
|
|
mock_runtime.return_value = mock_runtime
|
|
|
|
# test ordinary
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
sw.handle_keys(ord("h"))
|
|
|
|
mock_win_cmd.key_pub.publish.assert_called_once_with("h")
|
|
|
|
# test bad key
|
|
def bad_key(k):
|
|
raise ValueError("Bad Key")
|
|
|
|
mock_win_cmd.key_pub.publish = bad_key
|
|
|
|
sw.handle_keys(ord("b"))
|
|
|
|
mock_runtime.assert_called_once_with(
|
|
f"Unknown key code: [{ord('b')}]. Please report to the displayarray issue page."
|
|
)
|
|
mock_warnings.warn.assert_called_once_with(mock_runtime)
|
|
|
|
# test exit key
|
|
assert sw.ESC_KEY_CODES == [27]
|
|
ret = sw.handle_keys(27)
|
|
|
|
mock_destroy.assert_called_once_with("displayarray (press ESC to quit)")
|
|
assert sw.exited is True
|
|
mock_win_cmd.quit.assert_called()
|
|
mock_stop.assert_called_with("0")
|
|
assert ret == "quit"
|
|
|
|
|
|
def test_handle_mouse():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch(
|
|
"displayarray.window.subscriber_windows.MouseEvent"
|
|
) as mock_mouse_event:
|
|
mock_mousey = mock_mouse_event.return_value = mock.MagicMock()
|
|
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
sw.handle_mouse(1, 2, 3, 4, 5)
|
|
mock_mouse_event.assert_called_once_with(1, 2, 3, 4, 5)
|
|
mock_win_cmd.mouse_pub.publish.assert_called_once_with(mock_mousey)
|
|
|
|
|
|
def test_update_frames():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
cv2, "imshow"
|
|
) as mock_imshow:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
frame = np.ones((100, 100))
|
|
sw.FRAME_DICT["0"] = frame
|
|
|
|
sw.update_frames()
|
|
|
|
assert sw.frames == {"0": [frame]}
|
|
mock_imshow.assert_called_once_with("displayarray (press ESC to quit)", frame)
|
|
|
|
|
|
def test_update_frames_callback():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
cv2, "imshow"
|
|
) as mock_imshow:
|
|
cb = mock.MagicMock()
|
|
cb2 = mock.MagicMock()
|
|
frame = np.ones((100, 100))
|
|
frame2 = np.ones((102, 102))
|
|
frame3 = np.ones((103, 103))
|
|
cb.return_value = frame2
|
|
cb2.return_value = frame3
|
|
|
|
sw = sub_win.SubscriberWindows(
|
|
window_names=["0", "1"], video_sources=[0, 1], callbacks=[cb, cb2]
|
|
)
|
|
|
|
sw.FRAME_DICT["0"] = frame
|
|
sw.FRAME_DICT["1"] = frame
|
|
|
|
sw.update_frames()
|
|
|
|
assert sw.frames == {"0": [frame3], "1": [frame3]}
|
|
assert np.all(cb.mock_calls[0].args[0] == frame)
|
|
assert np.all(cb2.mock_calls[0].args[0] == frame2)
|
|
mock_imshow.assert_has_calls(
|
|
[
|
|
mock.call("0 (press ESC to quit)", frame3),
|
|
mock.call("1 (press ESC to quit)", frame3),
|
|
]
|
|
)
|
|
|
|
|
|
def test_update_frames_too_many_channels():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
cv2, "imshow"
|
|
) as mock_imshow, mock.patch(
|
|
"displayarray.window.subscriber_windows.print"
|
|
) as mock_print:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
frame = np.ones((100, 100, 100))
|
|
sw.FRAME_DICT["0"] = frame
|
|
|
|
sw.update_frames()
|
|
|
|
mock_print.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
"Too many channels in output. (Got 100 instead of 1 or 3.) Frame selection callback added."
|
|
),
|
|
mock.call(
|
|
"Ctrl+scroll to change first channel.\n"
|
|
"Shift+scroll to change second channel.\n"
|
|
"Alt+scroll to change third channel."
|
|
),
|
|
]
|
|
)
|
|
|
|
assert isinstance(sw.callbacks[-1], SelectChannels)
|
|
assert sw.callbacks[-1].mouse_control is not None
|
|
assert sw.callbacks[-1].mouse_print_channels is True
|
|
assert sw.frames["0"][0].shape[-1] == 3
|
|
|
|
|
|
def test_update_frames_nested():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
cv2, "imshow"
|
|
) as mock_imshow, mock.patch("displayarray.window.subscriber_windows.print"):
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
frame = np.ones((20, 100, 100, 100))
|
|
sw.FRAME_DICT["0"] = frame
|
|
|
|
sw.update_frames()
|
|
|
|
assert np.all(sw.frames["0"][0] == np.ones((20, 100, 100, 3)))
|
|
assert len(sw.frames) == 1
|
|
assert mock_imshow.mock_calls[0].args[0] == "displayarray (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[0].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[1].args[0] == "0 - 1 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[1].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[2].args[0] == "0 - 2 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[2].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[3].args[0] == "0 - 3 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[3].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[4].args[0] == "0 - 4 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[4].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[5].args[0] == "0 - 5 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[5].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[6].args[0] == "0 - 6 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[6].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[7].args[0] == "0 - 7 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[7].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[8].args[0] == "0 - 8 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[8].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[9].args[0] == "0 - 9 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[9].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[10].args[0] == "0 - 10 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[10].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[11].args[0] == "0 - 11 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[11].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[12].args[0] == "0 - 12 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[12].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[13].args[0] == "0 - 13 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[13].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[14].args[0] == "0 - 14 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[14].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[15].args[0] == "0 - 15 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[15].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[16].args[0] == "0 - 16 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[16].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[17].args[0] == "0 - 17 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[17].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[18].args[0] == "0 - 18 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[18].args[1] == np.ones((100, 100, 3)))
|
|
assert mock_imshow.mock_calls[19].args[0] == "0 - 19 (press ESC to quit)"
|
|
assert np.all(mock_imshow.mock_calls[19].args[1] == np.ones((100, 100, 3)))
|
|
|
|
|
|
def test_update_frames_exception():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
cv2, "imshow"
|
|
) as mock_imshow:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
frame = RuntimeError("Sent from FrameUpdater")
|
|
sw.FRAME_DICT["0"] = frame
|
|
|
|
with pytest.raises(RuntimeError) as e:
|
|
sw.update_frames()
|
|
assert e.value == frame
|
|
|
|
|
|
def test_update():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "update_frames"
|
|
) as mock_update_win_frames, mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch.object(
|
|
sub_win.SubscriberWindows, "handle_keys"
|
|
) as mock_handle_keys, mock.patch.object(
|
|
cv2, "waitKey"
|
|
) as key:
|
|
sub_cmd = mock_win_cmd.win_cmd_sub.return_value = mock.MagicMock()
|
|
key.return_value = 2
|
|
mock_cmd = sub_cmd.get.return_value = mock.MagicMock()
|
|
mock_key = mock_handle_keys.return_value = mock.MagicMock()
|
|
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
cmd, key = sw.update()
|
|
|
|
assert mock_win_cmd.win_cmd_sub.call_count == 2
|
|
assert mock_update_win_frames.call_count == 2
|
|
assert sub_cmd.get.call_count == 2
|
|
assert cmd == mock_cmd
|
|
assert key == mock_key
|
|
|
|
|
|
|
|
def test_wait_for_init():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "update"
|
|
) as update:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
def mock_update():
|
|
sw.frames = mock_update.frames[mock_update.i]
|
|
mock_update.i += 1
|
|
return "", ""
|
|
|
|
mock_update.frames = [[], [], [], [1]]
|
|
mock_update.i = 0
|
|
|
|
update.side_effect = mock_update
|
|
|
|
sw.wait_for_init()
|
|
|
|
assert mock_update.i == 4
|
|
|
|
|
|
def test_end():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch(
|
|
"displayarray.window.subscriber_windows.subscriber_dictionary.stop_cam"
|
|
) as mock_stop:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
sw.close_threads = [mock.MagicMock(), mock.MagicMock()]
|
|
|
|
sw.end()
|
|
|
|
mock_win_cmd.quit.assert_called_with(force_all_read=False)
|
|
mock_stop.assert_called_with("0")
|
|
sw.close_threads[0].join.assert_called_once()
|
|
sw.close_threads[1].join.assert_called_once()
|
|
|
|
|
|
def test_enter_exit():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "end"
|
|
) as end:
|
|
with sub_win.SubscriberWindows() as sw:
|
|
assert isinstance(sw, sub_win.SubscriberWindows)
|
|
|
|
end.assert_called()
|
|
|
|
|
|
def test_del():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "end"
|
|
) as end:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
del sw
|
|
|
|
end.assert_called_once()
|
|
|
|
|
|
def test_loop():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch(
|
|
"displayarray.window.subscriber_windows.subscriber_dictionary.stop_cam"
|
|
) as mock_stop, mock.patch.object(
|
|
sub_win.SubscriberWindows, "update"
|
|
) as update:
|
|
sub_cmd = mock_win_cmd.win_cmd_sub.return_value = mock.MagicMock()
|
|
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
def mock_update():
|
|
mock_update.i += 1
|
|
return "", mock_update.keys[mock_update.i]
|
|
|
|
mock_update.keys = ["", "", "", "quit"]
|
|
mock_update.i = 0
|
|
|
|
update.side_effect = mock_update
|
|
|
|
sw.loop()
|
|
|
|
assert mock_update.i == 3
|
|
sub_cmd.release.assert_called_once()
|
|
mock_win_cmd.quit.assert_called_with(force_all_read=False)
|
|
mock_stop.assert_called_with("0")
|
|
|
|
|
|
def test_display():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch(
|
|
"displayarray.window.subscriber_windows.FrameUpdater"
|
|
) as fup, mock.patch(
|
|
"displayarray.window.subscriber_windows.SubscriberWindows"
|
|
) as sws:
|
|
fup_inst = fup.return_value = mock.MagicMock()
|
|
sws_inst = sws.return_value = mock.MagicMock()
|
|
|
|
d = sub_win.display(0, 1, size=(50, 50))
|
|
|
|
fup.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
0, force_backend="", fps_limit=float("inf"), request_size=(50, 50)
|
|
),
|
|
mock.call(
|
|
1, force_backend="", fps_limit=float("inf"), request_size=(50, 50)
|
|
),
|
|
]
|
|
)
|
|
assert fup_inst.start.call_count == 2
|
|
sws.assert_called_once_with(
|
|
window_names=["window 0", "window 1"], video_sources=(0, 1), silent=False
|
|
)
|
|
assert sws_inst.close_threads == [fup_inst, fup_inst]
|
|
assert d == sws_inst
|
|
|
|
def test_display_callbacks():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch(
|
|
"displayarray.window.subscriber_windows.FrameUpdater"
|
|
) as fup, mock.patch(
|
|
"displayarray.window.subscriber_windows.SubscriberWindows"
|
|
) as sws:
|
|
cb = mock.MagicMock()
|
|
|
|
sub_win.display(0, 1, callbacks=cb)
|
|
|
|
fup.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
0,
|
|
callbacks=[cb],
|
|
force_backend="",
|
|
fps_limit=float("inf"),
|
|
request_size=(-1, -1),
|
|
),
|
|
mock.call(
|
|
1,
|
|
callbacks=[cb],
|
|
force_backend="",
|
|
fps_limit=float("inf"),
|
|
request_size=(-1, -1),
|
|
),
|
|
]
|
|
)
|
|
|
|
fup.reset_mock()
|
|
|
|
cb2 = mock.MagicMock()
|
|
|
|
sub_win.display(0, 1, callbacks=[cb, cb2], fps_limit=60)
|
|
|
|
fup.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
0,
|
|
callbacks=[cb, cb2],
|
|
force_backend="",
|
|
fps_limit=60,
|
|
request_size=(-1, -1),
|
|
),
|
|
mock.call(
|
|
1,
|
|
callbacks=[cb, cb2],
|
|
force_backend="",
|
|
fps_limit=60,
|
|
request_size=(-1, -1),
|
|
),
|
|
]
|
|
)
|
|
|
|
'''
|
|
|
|
|
|
def test_add_window():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow") as mock_named_window, mock.patch.object(
|
|
cv2, "setMouseCallback"
|
|
) as mock_set_mouse, mock.patch(
|
|
"displayarray.window.subscriber_windows.WeakMethod", new_callable=mock.MagicMock
|
|
) as mock_weak:
|
|
weak_method = mock_weak.return_value = mock.MagicMock()
|
|
|
|
sw = sub_win.SubscriberWindows().add_window("second window")
|
|
|
|
mock_weak.assert_has_calls(
|
|
[mock.call(sw.handle_mouse), mock.call(sw.handle_mouse)]
|
|
)
|
|
assert sw.window_names == ["displayarray", "second window"]
|
|
mock_named_window.assert_has_calls(
|
|
[
|
|
mock.call("displayarray (press ESC to quit)"),
|
|
mock.call("second window (press ESC to quit)"),
|
|
]
|
|
)
|
|
mock_set_mouse.assert_has_calls(
|
|
[
|
|
mock.call("displayarray (press ESC to quit)", weak_method),
|
|
mock.call("second window (press ESC to quit)", weak_method),
|
|
]
|
|
)
|
|
|
|
|
|
def test_update_with_array():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch.object(cv2, "namedWindow"), mock.patch.object(
|
|
sub_win.SubscriberWindows, "update_frames"
|
|
) as mock_update_win_frames, mock.patch(
|
|
"displayarray.window.subscriber_windows.window_commands"
|
|
) as mock_win_cmd, mock.patch.object(
|
|
sub_win.SubscriberWindows, "handle_keys"
|
|
) as mock_handle_keys, mock.patch.object(
|
|
sub_win.SubscriberWindows, "add_source"
|
|
) as add_source, mock.patch.object(
|
|
sub_win.SubscriberWindows, "add_window"
|
|
) as add_window, mock.patch(
|
|
"displayarray.window.subscriber_windows.global_cv_display_callback"
|
|
) as mock_cb, mock.patch.object(
|
|
cv2, "waitKey"
|
|
) as key:
|
|
sw = sub_win.SubscriberWindows()
|
|
|
|
sw.update(arr=1, id=2)
|
|
|
|
mock_cb.assert_called_once_with(1, 2)
|
|
add_source.assert_has_calls([mock.call(0), mock.call(2)])
|
|
add_window.assert_has_calls([mock.call("displayarray"), mock.call(2)])
|
|
|
|
|
|
def test_display_blocking():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch(
|
|
"displayarray.window.subscriber_windows.FrameUpdater"
|
|
) as fup, mock.patch(
|
|
"displayarray.window.subscriber_windows.SubscriberWindows"
|
|
) as sws:
|
|
fup_inst = fup.return_value = mock.MagicMock()
|
|
sws_inst = sws.return_value = mock.MagicMock()
|
|
|
|
sub_win.display(0, 1, blocking=True)
|
|
|
|
assert fup_inst.start.call_count == 2
|
|
sws.assert_called_once_with(
|
|
window_names=["window 0", "window 1"], video_sources=(0, 1), silent=False
|
|
)
|
|
sws_inst.loop.assert_called_once()
|
|
assert fup_inst.join.call_count == 2
|
|
|
|
|
|
def test_display_callbacks_dict():
|
|
sub_win.SubscriberWindows.FRAME_DICT = {}
|
|
with mock.patch(
|
|
"displayarray.window.subscriber_windows.FrameUpdater"
|
|
) as fup, mock.patch(
|
|
"displayarray.window.subscriber_windows.SubscriberWindows"
|
|
) as sws:
|
|
cb1 = mock.MagicMock()
|
|
cb2 = mock.MagicMock()
|
|
cb3 = mock.MagicMock()
|
|
|
|
sub_win.display(0, 1, 2, callbacks={0: cb1, 1: [cb1, cb2], "2": [cb3]})
|
|
|
|
fup.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
0,
|
|
callbacks=[cb1],
|
|
force_backend="",
|
|
fps_limit=float("inf"),
|
|
request_size=(-1, -1),
|
|
),
|
|
mock.call(
|
|
1,
|
|
callbacks=[cb1, cb2],
|
|
force_backend="",
|
|
fps_limit=float("inf"),
|
|
request_size=(-1, -1),
|
|
),
|
|
mock.call(
|
|
2,
|
|
callbacks=[cb3],
|
|
force_backend="",
|
|
fps_limit=float("inf"),
|
|
request_size=(-1, -1),
|
|
),
|
|
]
|
|
)
|