Added test for batch widget

.github/workflows/coveralls.yml

@@ -2,9 +2,9 @@ name: coveralls
 
 on:
   push:
-    branches: [master, development]
+    branches: [ master, development ]
   pull_request:
-    branches: [master, development]
+    branches: [ master, development ]
 
 jobs:
   coveralls:

src/asammdf/gui/widgets/batch.py

@@ -415,6 +415,9 @@ def extract_bus_logging_thread(self, source_files, database_files, count, compre
                 overwrite=True,
                 progress=progress,
             )
+
+            mdf.close()
+
             if result is TERMINATED:
                 return
 
@@ -611,6 +614,9 @@ def extract_bus_csv_logging_thread(
                 add_units=add_units,
                 progress=progress,
             )
+
+            mdf.close()
+
             if result is TERMINATED:
                 return
 
@@ -705,7 +711,7 @@ def concatenate(self, event=None):
         self._progress = setup_progress(parent=self, autoclose=False)
         self._progress.finished.connect(self.concatenate_finished)
 
-        self._progress.run_thread_with_progress(
+        rez = self._progress.run_thread_with_progress(
             target=self.concatenate_thread,
             args=(
                 output_file_name,
@@ -747,6 +753,9 @@ def concatenate_thread(
             progress=progress,
         )
 
+        for file in files:
+            file.close()
+
         if result is TERMINATED:
             return
         else:
@@ -764,6 +773,8 @@ def concatenate_thread(
             progress=progress,
         )
 
+        mdf.close()
+
         if result is not TERMINATED:
             return result
 
@@ -812,6 +823,11 @@ def stack_thread(
             progress=progress,
         )
 
+        for file in files:
+            file.close()
+
+        mdf.close()
+
         if result is not TERMINATED:
             return result
 
@@ -1469,6 +1485,7 @@ def apply_processing_thread(self, progress):
                 if result is TERMINATED:
                     return
                 else:
+                    mdf.close()
                     mdf = result
 
                 mdf.configure(
@@ -1590,8 +1607,6 @@ def apply_processing_thread(self, progress):
                     else:
                         file_name = output_folder / Path(mdf_file.original_name).relative_to(root)
 
-                    print(file_name)
-
                     if not file_name.parent.exists():
                         os.makedirs(file_name.parent, exist_ok=True)
                 else:
@@ -1614,6 +1629,8 @@ def apply_processing_thread(self, progress):
                     progress=progress,
                 )
 
+                mdf.close()
+
                 if result is TERMINATED:
                     return
 
@@ -1674,6 +1691,8 @@ def apply_processing_thread(self, progress):
 
                 target(**kwargs)
 
+            mdf.close()
+
     def change_modify_output_folder(self, event=None):
         folder = QtWidgets.QFileDialog.getExistingDirectory(self, "Select output folder", "")
         if folder:
@@ -1933,6 +1952,8 @@ def load_filter_list(self, event=None, file_name=None):
 
                 self.update_selected_filter_channels()
 
+                mdf.close()
+
     def connect_export_updates(self):
         self.output_format.currentTextChanged.connect(self.store_export_setttings)
 

test/asammdf/gui/test_base.py

@@ -20,6 +20,7 @@ class DragAndDrop
 import unittest
 from unittest import mock
 
+from h5py import File as HDF5
 import pyqtgraph
 from PySide6 import QtCore, QtGui, QtTest, QtWidgets
 
@@ -97,7 +98,10 @@ def processEvents(timeout=0.001):
 
     def setUp(self) -> None:
         if os.path.exists(self.test_workspace):
-            shutil.rmtree(self.test_workspace)
+            try:
+                shutil.rmtree(self.test_workspace)
+            except PermissionError as e:
+                print(e)
         if not os.path.exists(self.screenshots):
             os.makedirs(self.screenshots)
 
@@ -459,18 +463,48 @@ def search_y_of_signal_in_column(pixmap_column, signal_color):
         return line
 
 
-class OpenMDF:
-    def __init__(self, file_path):
-        self.mdf = None
+class OpenFileContextManager:
+    """
+    Generic class for opening a file using context manager.
+    Methods:
+        __enter__: return opened file object.
+        __exit__: close file object. If exc_type, exc_val, exc_tb, raise exception.
+    """
+
+    def __init__(self, file_path: str | pathlib.Path):
+        """
+        Parameters
+        ----------
+        file_path: file path as str or pathlib.Path object
+        """
+        self.file = None
         self._file_path = file_path
-        self._process_bus_logging = ("process_bus_logging", True)
 
     def __enter__(self):
-        self.mdf = mdf.MDF(self._file_path, process_bus_logging=self._process_bus_logging)
-        return self.mdf
+        pass
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+        self.file.close()
         for exc in (exc_type, exc_val, exc_tb):
             if exc is not None:
                 raise exc
-        self.mdf.close()
+
+
+class OpenMDF(OpenFileContextManager):
+    """
+    Open MDF file using context manager.
+    """
+
+    def __enter__(self):
+        self.file = mdf.MDF(self._file_path, process_bus_logging=("process_bus_logging", True))
+        return self.file
+
+
+class OpenHDF5(OpenFileContextManager):
+    """
+    Open HDF5 file using context manager.
+    """
+
+    def __enter__(self):
+        self.file = HDF5(self._file_path)
+        return self.file

test/asammdf/gui/widgets/batch/test_BatchWidget_FilesList.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+from pathlib import Path
+
+from PySide6 import QtCore, QtTest
+
+from test.asammdf.gui.widgets.test_BaseBatchWidget import TestBatchWidget
+
+# Note: If it's possible and make sense, use self.subTests
+# to avoid initializing widgets multiple times and consume time.
+
+
+class TestPushButtons(TestBatchWidget):
+    test_file = "test_batch.mf4"
+
+    def test_PushButtons_Sort(self):
+        """
+        Events:
+            - Open 'BatchWidget' with 2 valid measurements.
+            - Press PushButton "Sort by start time"
+            - Press PushButton "Sort alphabetically"
+
+        Evaluate:
+            - Files are sorted by start time
+            - Files are sorted alphabetically
+        """
+        # Setup
+        file_0 = Path(self.test_workspace, self.default_test_file)
+        file_1 = Path(self.test_workspace, self.test_file)
+
+        alpha_sort = [str(file_0), str(file_1)]
+        time_sort = [str(file_1), str(file_0)]
+
+        self.copy_mdf_files_to_workspace()
+        self.setUpBatchWidget(measurement_files=alpha_sort)
+
+        # Event
+        with self.subTest("test sort by start time btn"):
+            # Press `Sort by start time` button
+            QtTest.QTest.mouseClick(self.widget.sort_by_start_time_btn, QtCore.Qt.MouseButton.LeftButton)
+
+            # Evaluate
+            self.assertEqual([self.widget.files_list.item(row).text() for row in range(2)], time_sort)
+
+        with self.subTest("test sort alphabetically btn"):
+            # Press `Sort by start time` button
+            QtTest.QTest.mouseClick(self.widget.sort_alphabetically_btn, QtCore.Qt.MouseButton.LeftButton)
+
+            # Evaluate
+            self.assertEqual([self.widget.files_list.item(row).text() for row in range(2)], alpha_sort)

test/asammdf/gui/widgets/batch/test_BatchWidget_Tab_BusLogging.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+
+from test.asammdf.gui.widgets.test_BaseBatchWidget import TestBatchWidget
+
+# Note: If it's possible and make sense, use self.subTests
+# to avoid initializing widgets multiple times and consume time.
+
+
+class TestPushButtons(TestBatchWidget):
+    default_test_file = "ASAP2_Demo_V171.mf4"
+    class_test_file = "test_batch.mf4"

test/asammdf/gui/widgets/batch/test_BatchWidget_Tab_Concatenate.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python
+from pathlib import Path
+from unittest import mock
+
+from PySide6 import QtCore, QtTest
+
+from test.asammdf.gui.test_base import OpenMDF
+from test.asammdf.gui.widgets.test_BaseBatchWidget import TestBatchWidget
+
+# Note: If it's possible and make sense, use self.subTests
+# to avoid initializing widgets multiple times and consume time.
+
+
+class TestPushButtons(TestBatchWidget):
+    test_file_0_name = "test_batch_cut_0.mf4"
+    test_file_1_name = "test_batch_cut_1.mf4"
+    output_file_name = "output.mf4"
+
+    def setUp(self):
+        super().setUp()
+
+        self.test_file_0 = Path(self.test_workspace, self.test_file_0_name)
+        self.test_file_1 = Path(self.test_workspace, self.test_file_1_name)
+
+        self.copy_mdf_files_to_workspace()
+        self.setUpBatchWidget(measurement_files=[str(self.test_file_0), str(self.test_file_1)])
+
+        # Go to Tab: Concatenate
+        self.widget.aspects.setCurrentIndex(self.concatenate_aspect)
+        self.processEvents(0.1)
+
+    def test_PushButton_Concatenate(self):
+        """
+        Events:
+            - Open 'BatchWidget' with created measurement.
+            - Go to Tab: "Concatenate": Index 0
+            - Press PushButton "Concatenate"
+
+        Evaluate:
+            - New file is created
+            - No channel from first file is found in 2nd file (scrambled file)
+        """
+        # Get evaluation data
+        output_file = Path(self.test_workspace, self.output_file_name)
+
+        channels = set()
+        with OpenMDF(self.test_file_0) as mdf_file:
+            for channel in mdf_file.iter_channels():
+                channels.add(channel.name)
+            expected_min = channel.timestamps.min()
+
+        with OpenMDF(self.test_file_1) as mdf_file:
+            channel = next(mdf_file.iter_channels())
+            expected_max = channel.timestamps.max()
+
+        # Event
+        with mock.patch("asammdf.gui.widgets.batch.QtWidgets.QFileDialog.getSaveFileName") as mo_getSaveFileName:
+            mo_getSaveFileName.return_value = output_file, ""
+            QtTest.QTest.mouseClick(self.widget.concatenate_btn, QtCore.Qt.MouseButton.LeftButton)
+        # Allow progress bar to close
+        self.processEvents(2)
+
+        # Evaluate that file exist
+        self.assertTrue(output_file.exists())
+
+        # Evaluate
+        with OpenMDF(output_file) as mdf_file:
+            # Evaluate saved file
+            for name in channels:
+                self.assertIn(name, mdf_file.channels_db.keys())
+
+            channel = next(mdf_file.iter_channels())
+            self.assertAlmostEqual(channel.timestamps.min(), expected_min)
+            self.assertAlmostEqual(channel.timestamps.max(), expected_max)