From 2925a5f20ef3b0dc6d66f2d9781775f53c01dbfc Mon Sep 17 00:00:00 2001 From: wyzula-jan <133381102+wyzula-jan@users.noreply.github.com> Date: Tue, 17 Oct 2023 13:40:30 +0200 Subject: [PATCH] test: test_stream_plot.py basic tests for stream_plot.py, test_basic_plot.py removed --- .../examples/stream_plot/stream_plot.py | 29 ---- tests/test_basic_plot.py | 152 ---------------- tests/test_stream_plot.py | 162 ++++++++++++++++++ 3 files changed, 162 insertions(+), 181 deletions(-) delete mode 100644 tests/test_basic_plot.py create mode 100644 tests/test_stream_plot.py diff --git a/bec_widgets/examples/stream_plot/stream_plot.py b/bec_widgets/examples/stream_plot/stream_plot.py index c45a8bfc..665cc04b 100644 --- a/bec_widgets/examples/stream_plot/stream_plot.py +++ b/bec_widgets/examples/stream_plot/stream_plot.py @@ -67,7 +67,6 @@ class StreamPlot(QtWidgets.QWidget): self.init_ui() self.init_curves() self.hook_crosshair() - self.pushButton_generate.clicked.connect(self.generate_data) def init_ui(self): """Setup all ui elements""" @@ -201,34 +200,6 @@ class StreamPlot(QtWidgets.QWidget): # ROI self.roi_selector.sigRegionChangeFinished.connect(self.get_roi_region) - def generate_data(self): - def gauss(x, mu, sigma): - return (1 / (sigma * np.sqrt(2 * np.pi))) * np.exp(-0.5 * ((x - mu) / sigma) ** 2) - - self.plotter_data_x = np.linspace(0, 10, 1000) - self.plotter_data_y = [ - gauss(self.plotter_data_x, 1, 1), - gauss(self.plotter_data_x, 1.5, 3), - np.sin(self.plotter_data_x), - np.cos(self.plotter_data_x), - np.sin(2 * self.plotter_data_x), - ] # List of y-values for multiple curves - self.y_value_list = ["Gauss (1,1)", "Gauss (1.5,3)"] # ["Sine"]#, "Cosine", "Sine2x"] - - # Curves - color_list = ["#384c6b", "#e28a2b", "#5E3023", "#e41a1c", "#984e83", "#4daf4a"] - - self.init_curves() - - for ii in range(len(self.y_value_list)): - self.curves[ii].setData(self.plotter_data_x, self.plotter_data_y[ii]) - - self.data_2D = np.random.random((150, 30)) - self.img.setImage(self.data_2D) - - if self.roi_selector not in self.plot.items: - self.plot.addItem(self.roi_selector) - def get_roi_region(self): """For testing purpose now, get roi region and print it to self.label as tuple""" region = self.roi_selector.getRegion() diff --git a/tests/test_basic_plot.py b/tests/test_basic_plot.py deleted file mode 100644 index ba4bb154..00000000 --- a/tests/test_basic_plot.py +++ /dev/null @@ -1,152 +0,0 @@ -# from unittest import mock -# -# import numpy as np -# from pytestqt import qtbot -# -# from bec_widgets import line_plot -# -# -# def test_line_plot_emits_no_signal(qtbot): -# """Test LinePlot emits no signal when only one data entry is present.""" -# -# y_value_list = ["y1", "y2"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# data = { -# "data": { -# "x": {"x": {"value": 1}}, -# "y1": {"y1": {"value": 1}}, -# "y2": {"y2": {"value": 3}}, -# } -# } -# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]} -# with mock.patch("bec_widgets.line_plot.client") as mock_client: -# with mock.patch.object(plot, "update_signal") as mock_update_signal: -# plot.on_scan_segment(data=data, metadata=metadata) -# mock_update_signal.emit.assert_not_called() -# -# -# def test_line_plot_emits_signal(qtbot): -# """Test LinePlot emits signal.""" -# -# y_value_list = ["y1", "y2"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# data = { -# "data": { -# "x": {"x": {"value": 1}}, -# "y1": {"y1": {"value": 1}}, -# "y2": {"y2": {"value": 3}}, -# } -# } -# plotter_data_y = [[1, 1], [3, 3]] -# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]} -# with mock.patch("bec_widgets.line_plot.client") as mock_client: -# # mock_client.device_manager.devices.keys.return_value = ["y1"] -# with mock.patch.object(plot, "update_signal") as mock_update_signal: -# mock_update_signal.emit() -# plot.on_scan_segment(data=data, metadata=metadata) -# plot.on_scan_segment(data=data, metadata=metadata) -# mock_update_signal.emit.assert_called() -# # TODO allow mock_client to create return values for device_manager_devices -# # assert plot.plotter_data_y == plotter_data_y -# -# -# def test_line_plot_raise_warning_wrong_signal_request(qtbot): -# """Test LinePlot raises warning and skips signal when entry not present in data.""" -# -# y_value_list = ["y1", "y22"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# data = { -# "data": { -# "x": {"x": {"value": [1, 2, 3, 4, 5]}}, -# "y1": {"y1": {"value": [1, 2, 3, 4, 5]}}, -# "y2": {"y2": {"value": [1, 2, 3, 4, 5]}}, -# } -# } -# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]} -# with mock.patch("bec_widgets.line_plot.client") as mock_client: -# # TODO fix mock_client -# mock_dict = {"y1": [1, 2]} -# mock_client.device_manager.devices.__contains__.side_effect = mock_dict.__contains__ -# -# # = {"y1": [1, 2]} -# with mock.patch.object(plot, "update_signal") as mock_update_signal: -# mock_update_signal.emit() -# plot.on_scan_segment(data=data, metadata=metadata) -# assert plot.y_value_list == ["y1"] - - -# def test_line_plot_update(qtbot): -# """Test LinePlot update.""" - -# y_value_list = ["y1", "y2"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# plot.label_bottom = "x" -# plot.label_left = f"{', '.join(y_value_list)}" -# plot.plotter_data_x = [1, 2, 3, 4, 5] -# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]] -# plot.update() - -# assert all(plot.curves[0].getData()[0] == np.array([1, 2, 3, 4, 5])) -# assert all(plot.curves[0].getData()[1] == np.array([1, 2, 3, 4, 5])) -# assert all(plot.curves[1].getData()[1] == np.array([3, 4, 5, 6, 7])) - - -# # TODO Outputting the wrong data, e.g. motor is not in list of devices -# def test_line_plot_update(qtbot): -# """Test LinePlot update.""" - -# y_value_list = ["y1", "y2"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# plot.label_bottom = "x" -# plot.label_left = f"{', '.join(y_value_list)}" -# plot.plotter_data_x = [1, 2, 3, 4, 5] -# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]] -# plot.update() - -# assert all(plot.curves[0].getData()[0] == np.array([1, 2, 3, 4, 5])) -# assert all(plot.curves[0].getData()[1] == np.array([1, 2, 3, 4, 5])) -# assert all(plot.curves[1].getData()[1] == np.array([3, 4, 5, 6, 7])) - - -# def test_line_plot_mouse_moved(qtbot): -# """Test LinePlot mouse_moved.""" - -# y_value_list = ["y1", "y2"] -# plot = line_plot.BasicPlot(y_value_list=y_value_list) -# plot.plotter_data_x = [1, 2, 3, 4, 5] -# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]] -# plot.precision = 3 -# string_cap = 10 -# x_data = f"{3:.{plot.precision}f}" -# y_data = f"{3:.{plot.precision}f}" -# output_string = "".join( -# [ -# "Mouse cursor", -# "\n", -# f"{y_value_list[0]}", -# "\n", -# f"X_data: {x_data:>{string_cap}}", -# "\n", -# f"Y_data: {y_data:>{string_cap}}", -# ] -# ) -# x_data = f"{3:.{plot.precision}f}" -# y_data = f"{5:.{plot.precision}f}" -# output_string = "".join( -# [ -# output_string, -# "\n", -# f"{y_value_list[1]}", -# "\n", -# f"X_data: {x_data:>{string_cap}}", -# "\n", -# f"Y_data: {y_data:>{string_cap}}", -# ] -# ) -# with mock.patch.object( -# plot, "plot" -# ) as mock_plot: # TODO change test to simulate QTable instead of QLabel -# mock_plot.sceneBoundingRect.contains.return_value = True -# mock_plot.vb.mapSceneToView((20, 10)).x.return_value = 2.8 -# plot.mouse_moved((20, 10)) -# assert plot.mouse_box_data.text() == output_string diff --git a/tests/test_stream_plot.py b/tests/test_stream_plot.py new file mode 100644 index 00000000..f74f183e --- /dev/null +++ b/tests/test_stream_plot.py @@ -0,0 +1,162 @@ +from unittest import mock + +import numpy as np +import pytest +from bec_lib.core import BECMessage +from pytestqt import qtbot +import threading + +from bec_lib.core import RedisConnector +from bec_widgets.examples.stream_plot.stream_plot import StreamPlot + + +@pytest.fixture +def stream_app(qtbot): + """Helper function to set up the StreamPlot widget.""" + client = mock.MagicMock() + widget = StreamPlot(client=client) + qtbot.addWidget(widget) + qtbot.waitExposed(widget) + return widget + + +def test_roi_signals_emitted(qtbot, stream_app): + region = (0.1, 0.9) + with qtbot.waitSignal(stream_app.roi_signal, timeout=1000) as blocker: + stream_app.roi_signal.emit(region) + assert blocker.signal_triggered + assert blocker.args == [region] + + +def test_update_signals_emitted(qtbot, stream_app): + # Mimic data coming from the data stream + stream_app.plotter_data_x = [list(range(10))] # Replace with the actual x data + stream_app.plotter_data_y = [list(range(10))] # Replace with the actual y data + + # Initialize curves + stream_app.init_curves() + + with qtbot.waitSignal(stream_app.update_signal, timeout=1000) as blocker: + stream_app.update_signal.emit() + assert blocker.signal_triggered + + +def test_ui_initialization(qtbot, stream_app): + """Checking the UI creation.""" + + # Check if UI elements are initialized correctly + assert stream_app.label_plot is not None + assert stream_app.label_plot_moved is not None + assert stream_app.label_plot_clicked is not None + assert stream_app.label_image_moved is not None + assert stream_app.label_image_clicked is not None + + # Check if plots are initialized correctly + assert stream_app.plot is not None + assert stream_app.plot_image is not None + + # Check if ROI selector is initialized correctly + assert stream_app.roi_selector is not None + + +def test_1d_plotting_data(qtbot, stream_app): + # Set up some mock data + x_data = [list(range(10))] + y_data = [list(range(10))] + + # Manually set the data attributes + stream_app.plotter_data_x = x_data + stream_app.plotter_data_y = y_data + stream_app.y_value_list = ["Curve 1"] + + # Initialize curves and update the plot + stream_app.init_curves() + stream_app.update() # This should update the plot with the new data + + # Check the data on the plot + for idx, curve in enumerate(stream_app.curves): + np.testing.assert_array_equal(curve.xData, x_data[0]) # Access the first list of x_data + np.testing.assert_array_equal( + curve.yData, y_data[idx] + ) # Access the list of y_data for each curve without additional indexing + + +def test_flip_even_rows(qtbot, stream_app): + # Create a numpy array with some known data + original_array = np.array( + [ + [1, 2, 3, 4, 5], + [6, 7, 8, 9, 10], + [11, 12, 13, 14, 15], + [16, 17, 18, 19, 20], + ] + ) + + # Call flip_even_rows on the original array + flipped_array = stream_app.flip_even_rows(original_array) + + # Expected array flipped along the rows with even indices + expected_array = np.array( + [ + [1, 2, 3, 4, 5], + [10, 9, 8, 7, 6], + [11, 12, 13, 14, 15], + [20, 19, 18, 17, 16], + ] + ) + + # Check that flip_even_rows returned the expected result + np.testing.assert_array_equal(flipped_array, expected_array) + + +def test_on_dap_update(qtbot, stream_app): + """2D image rendering by dap update""" + # Create some mock data to be "received" by the slot + data_dict = {"data": {"z": np.random.rand(10, 10)}} + metadata_dict = {} + + # Trigger the slot + stream_app.on_dap_update(data_dict, metadata_dict) + + # Apply the same transformation to the test data + expected_data = stream_app.flip_even_rows(data_dict["data"]["z"]) + + # Now check the state of the StreamPlot object + # For example, check the data of the image plot: + np.testing.assert_array_equal(stream_app.img.image, expected_data) + + +# def test_new_proj(qtbot, stream_app): #TODO this test is not working, does it make sense testing even? +# # Create some mock content to be "received" by the slot +# content_dict = {"signals": {"proj_nr": 1}} +# metadata_dict = {} +# +# # Manually create some mock data that new_proj would work with +# # This step may need to be adjusted to fit the actual behavior of new_proj +# mock_data = { +# "q": np.array([1, 2, 3, 4, 5]), +# "norm_sum": np.array([6, 7, 8, 9, 10]), +# "metadata": "some_metadata", +# } +# +# # Assume the RedisConnector client would return this data when new_proj is called +# mock_message = mock.MagicMock(spec=BECMessage.DeviceMessage) +# mock_message.__getitem__.side_effect = lambda key: mock_data[key] +# stream_app.client.producer.get = mock.MagicMock(return_value=mock_message.dumps()) +# +# # Trigger the slot +# stream_app.new_proj(content_dict, metadata_dict) +# +# # Now check the state of the StreamPlot object +# # For example, check that the plotter_data_x attribute was updated correctly: +# np.testing.assert_array_equal(stream_app.plotter_data_x, [mock_data["q"]]) +# assert stream_app._current_proj == 1 +# assert stream_app._current_q == mock_data["q"] +# assert stream_app._current_norm == mock_data["norm_sum"] +# assert stream_app._current_metadata == mock_data["metadata"] + + +# def test_connection_creation(qtbot, stream_app): #TODO maybe test connections in a different way? +# assert isinstance(stream_app.producer, RedisConnector) +# assert isinstance(stream_app.data_retriever, threading.Thread) +# assert stream_app.data_retriever.is_alive()