some new tests

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

@@ -4,6 +4,7 @@
 
 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
@@ -43,12 +44,12 @@ def test_PushButton_Concatenate(self):
         output_file = Path(self.test_workspace, self.output_file_name)
 
         channels = set()
-        with self.OpenMDF(self.test_file_0) as mdf_file:
+        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 self.OpenMDF(self.test_file_1) as mdf_file:
+        with OpenMDF(self.test_file_1) as mdf_file:
             channel = next(mdf_file.iter_channels())
             expected_max = channel.timestamps.max()
 
@@ -63,7 +64,7 @@ def test_PushButton_Concatenate(self):
         self.assertTrue(output_file.exists())
 
         # Evaluate
-        with self.OpenMDF(output_file) as mdf_file:
+        with OpenMDF(output_file) as mdf_file:
             # Evaluate saved file
             for name in channels:
                 self.assertIn(name, mdf_file.channels_db.keys())

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

@@ -4,12 +4,15 @@
 from random import randint
 from unittest import mock
 
+import scipy
 from h5py import File as HDF5
 import pandas as pd
 from PySide6 import QtCore, QtTest, QtWidgets
 
+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.
 
@@ -53,7 +56,7 @@ def test_ScrambleTexts(self):
         self.assertTrue(scrambled_filepath.exists())
 
         # get saved file as MDF
-        with self.OpenMDF(scrambled_filepath) as mdf_file:
+        with OpenMDF(scrambled_filepath) as mdf_file:
             # Evaluate file
             for name in channels:
                 self.assertNotIn(name, mdf_file.channels_db)
@@ -124,7 +127,7 @@ def test_output_format_MDF(self):
         self.assertTrue(saved_file.exists())
 
         # get saved file as MDF
-        with self.OpenMDF(saved_file) as mdf_file:
+        with OpenMDF(saved_file) as mdf_file:
             # Evaluate saved file
             for channel in self.selected_channels:
                 self.assertIn(channel, mdf_file.channels_db)
@@ -145,7 +148,7 @@ def test_output_format_ASC(self):
 
         # Expected results
         saved_file = Path(self.test_workspace, self.default_test_file.replace(".mf4", ".asc"))
-        with self.OpenMDF(self.measurement_file) as mdf_file:
+        with OpenMDF(self.measurement_file) as mdf_file:
             start = mdf_file.start_time.strftime("%a %b %d %I:%M:%S.%f %p %Y")
 
         expected_text = f"date {start}\nbase hex  timestamps absolute\nno internal events logged\n"
@@ -180,14 +183,19 @@ def test_output_format_CSV(self):
         """
         # Ensure output format
         self.widget.output_format.setCurrentText("CSV")
+        # uncheck all checkboxes
+        for checkbox in self.widget.CSV.findChildren(QtWidgets.QCheckBox):
+            if checkbox.isChecked():
+                self.mouseClick_CheckboxButton(checkbox)
 
+        self.processEvents()
         # Expected results
         groups = self.get_selected_groups(channels=self.selected_channels)
 
         # Event
         QtTest.QTest.mouseClick(self.widget.apply_btn, QtCore.Qt.MouseButton.LeftButton)
         # Wait for thread to finish
-        self.processEvents(2)
+        self.processEvents(3)
 
         # Evaluate
         for index, (group_name, channels_list) in enumerate(groups.items()):
@@ -197,13 +205,12 @@ def test_output_format_CSV(self):
                     ".mf4", f"{group_name.replace(group_name[:4], f".ChannelGroup_{index}").replace(" ", "_")}.csv"
                 ),
             )
-            self.assertTrue(csv_file.exists())
-
+            self.assertTrue(csv_file.exists(), csv_file)
             pandas_tab = pd.read_csv(csv_file)
             for channel in channels_list:
                 self.assertIn(channel, pandas_tab.columns)
 
-        # ToDo is necessary to evaluate dataframe values (for each signal, min, max, len)?
+        # ToDo is necessary to evaluate dataframe values (for each signal, min, max, len)
 
     def test_output_format_HDF5(self):
         """
@@ -242,27 +249,70 @@ def test_output_format_MAT(self):
         """
         When QThreads are running, event-loops needs to be processed.
         Events:
-            - Ensure that output format is MDF
+            - Ensure that output format is MAT
             - Press PushButton Apply.
 
         Evaluate:
             - File was created.
             - Ensure that output file has only selected channels
         """
+        groups = self.get_selected_groups(channels=self.selected_channels)
+
         # Ensure output format
         self.widget.output_format.setCurrentText("MAT")
 
         # Expected results
-        hdf5_path = Path(self.test_workspace, self.default_test_file.replace(".mf4", ".mat"))
+        mat_path = Path(self.test_workspace, self.default_test_file.replace(".mf4", ".mat"))
         groups = self.get_selected_groups(channels=self.selected_channels)
 
         # Mouse click on Apply button
         QtTest.QTest.mouseClick(self.widget.apply_btn, QtCore.Qt.MouseButton.LeftButton)
         # Wait for thread to finish
 
+        self.processEvents(2)
+
         # Evaluate
+        self.assertTrue(mat_path.exists())
+
+        mat_file = scipy.io.loadmat(str(mat_path))
+        to_replace = {" ", ".", "[", "]"}
+        for index, channels_list in enumerate(groups.values()):
+            for channel in channels_list:
+                for character in to_replace:
+                    channel = channel.replace(character, "_")
+                self.assertIn(f"DG{index}_{channel}", mat_file)
+
+    def test_output_format_Parquet(self):
+        """
+        When QThreads are running, event-loops needs to be processed.
+        Events:
+            - Ensure that output format is Parquet
+            - Press PushButton Apply.
+
+        Evaluate:
+            - File was created.
+            - Ensure that output file has only selected channels
+        """
+        groups = self.get_selected_groups(channels=self.selected_channels)
+
+        # Ensure output format
+        self.widget.output_format.setCurrentText("Parquet")
+
+        # Expected results
+        parquet_path = Path(self.test_workspace, self.default_test_file.replace(".mf4", ".parquet"))
+        groups = self.get_selected_groups(channels=self.selected_channels)
+
+        # Mouse click on Apply button
+        QtTest.QTest.mouseClick(self.widget.apply_btn, QtCore.Qt.MouseButton.LeftButton)
+        # Wait for thread to finish
+
         self.processEvents(2)
-        # self.assertTrue(hdf5_path.exists())
+
+        # Evaluate
+        self.assertTrue(parquet_path.exists())
+        pandas_tab = pd.read_parquet(parquet_path)
+        from_parquet = set(pandas_tab.columns)
+        self.assertSetEqual(set(self.selected_channels), from_parquet)
 
     def test_cut_checkbox_0(self):
         """
@@ -283,7 +333,7 @@ def test_cut_checkbox_0(self):
 
         output_file = Path(self.test_workspace, self.default_test_file)
 
-        with self.OpenMDF(self.measurement_file) as mdf_file:
+        with OpenMDF(self.measurement_file) as mdf_file:
             self.start_time = mdf_file.start_time.astimezone().replace(tzinfo=None)
             self.ts_min = min(
                 [
@@ -311,7 +361,7 @@ def test_cut_checkbox_0(self):
 
         # Evaluate
         output_file.exists()
-        with self.OpenMDF(output_file) as mdf_file:
+        with OpenMDF(output_file) as mdf_file:
             time_dif = expected_start_time - mdf_file.start_time
             self.assertEqual(time_dif.microseconds, 0)
             self.assertEqual(time_dif.seconds, 0)
@@ -336,7 +386,7 @@ def test_cut_checkbox_1(self):
         # Expected values
         output_file = Path(self.test_workspace, self.default_test_file)
 
-        with self.OpenMDF(self.measurement_file) as mdf_file:
+        with OpenMDF(self.measurement_file) as mdf_file:
             self.start_time = mdf_file.start_time
             self.ts_min = min(
                 [
@@ -361,7 +411,7 @@ def test_cut_checkbox_1(self):
 
         # Evaluate
         output_file.exists()
-        with self.OpenMDF(output_file) as mdf_file:
+        with OpenMDF(output_file) as mdf_file:
             self.assertEqual(self.start_time, mdf_file.start_time)
 
             for channel in mdf_file.iter_channels():

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

@@ -4,6 +4,7 @@
 
 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
@@ -44,10 +45,10 @@ def test_PushButton_Stack(self):
         # Expected values
         expected_channels = set()
 
-        with self.OpenMDF(self.measurement_file_1) as mdf_file:
+        with OpenMDF(self.measurement_file_1) as mdf_file:
             expected_channels |= set(mdf_file.channels_db)
 
-        with self.OpenMDF(self.measurement_file_2) as mdf_file:
+        with OpenMDF(self.measurement_file_2) as mdf_file:
             expected_channels |= set(mdf_file.channels_db)
 
         # Event
@@ -61,5 +62,5 @@ def test_PushButton_Stack(self):
         self.assertTrue(self.saved_file.exists())
 
         # Evaluate saved file
-        with self.OpenMDF(self.saved_file) as mdf_file:
+        with OpenMDF(self.saved_file) as mdf_file:
             self.assertTrue(len(expected_channels - set(mdf_file.channels_db)) == 0)

test/asammdf/gui/widgets/test_BaseBatchWidget.py

@@ -111,19 +111,3 @@ def get_selected_groups(self, channels: list) -> dict:
             iterator += 1
             count += 1
         return groups
-
-    class OpenMDF:
-        def __init__(self, file_path):
-            self.mdf = 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
-
-        def __exit__(self, exc_type, exc_val, exc_tb):
-            for exc in (exc_type, exc_val, exc_tb):
-                if exc is not None:
-                    raise exc
-            self.mdf.close()