unittest for reftek

tests/model/reftek_data/test_reftek.py

+from unittest import TestCase
+from pathlib import Path
+
+from sohstationviewer.model.reftek_data.reftek import RT130
+from sohstationviewer.model.general_data.general_data import \
+    ProcessingDataError
+
+
+TEST_DATA_DIR = Path(__file__).resolve().parent.parent.parent.joinpath(
+    'test_data')
+reftek_data = TEST_DATA_DIR.joinpath("RT130-sample")
+reftek_gap_data = TEST_DATA_DIR.joinpath("RT130-gap")
+
+
+class TestReftek(TestCase):
+    def test_path_not_exist(self):
+        # raise exception when path not exist
+        args = {
+            'data_type': 'RT130',
+            'folder': '_',
+            'rt130_waveform_data_req': False,
+            'on_unittest': True
+        }
+        with self.assertRaises(ProcessingDataError) as context:
+            RT130(**args)
+            self.assertEqual(
+                str(context.exception),
+                "Path '_' not exist"
+            )
+
+    def test_read_soh(self):
+        args = {
+            'data_type': 'RT130',
+            'folder': reftek_data,
+            'req_soh_chans': [],
+            'rt130_waveform_data_req': False,
+            'on_unittest': True
+        }
+        expected_soh = [
+            'SOH/Data Def', 'Battery Volt', 'DAS Temp', 'Backup Volt',
+            'Disk Usage1', 'Disk Usage2', 'Dump Called/Comp', 'GPS On/Off/Err',
+            'GPS Lk/Unlk', 'Clk Phase Err']
+        obj = RT130(**args)
+        self.assertEqual(obj.found_data_streams, [9])
+        self.assertEqual(obj.keys, [('92EB', '25')])
+        self.assertEqual(
+            list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
+            [])
+        self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
+        self.assertEqual(len(obj.log_data['TEXT']), 0)
+        self.assertEqual(list(obj.log_data[('92EB', '25')].keys()), ['SOH'])
+        self.assertEqual(len(obj.log_data[('92EB', '25')]['SOH']), 1)
+        self.assertEqual(
+            obj.log_data[('92EB', '25')]['SOH'][0][:100],
+            '\nState of Health  17:150:00:00:00:000   ST: 92EB'
+            '\n150:00:00:00 REF TEK 130'
+            '\r\n150:00:00:00 CPU SOFTWARE')
+        self.assertEqual(list(obj.soh_data.keys()), [('92EB', '25')])
+        self.assertEqual(list(obj.soh_data[('92EB', '25')].keys()),
+                         expected_soh)
+
+    def test_read_waveform(self):
+        args = {
+            'data_type': 'RT130',
+            'folder': reftek_data,
+            'req_soh_chans': [],
+            'req_wf_chans': [1],
+            'rt130_waveform_data_req': True,
+            'on_unittest': True
+        }
+        expected_waveform = ['DS1-1', 'DS1-2', 'DS1-3']
+        obj = RT130(**args)
+        self.assertEqual(obj.found_data_streams, [9, 1, 1])
+        self.assertEqual(obj.keys, [('92EB', '25')])
+        self.assertEqual(
+            list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
+            expected_waveform)
+
+        self.assertEqual(list(obj.waveform_data[('92EB', '25')].keys()),
+                         expected_waveform)
+        self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
+        self.assertIn('Event DS1',
+                      list(obj.soh_data[('92EB', '25')].keys()))
+
+    def test_read_mass_pos(self):
+        args = {
+            'data_type': 'RT130',
+            'folder': reftek_data,
+            'req_soh_chans': ['_'],
+            'include_mp123zne': True,
+            'rt130_waveform_data_req': False,
+            'on_unittest': True
+        }
+        expected_mass_pos = ['MassPos1', 'MassPos2', 'MassPos3']
+        obj = RT130(**args)
+        self.assertEqual(obj.found_data_streams, [9])
+        self.assertEqual(obj.keys, [('92EB', '25')])
+        self.assertEqual(
+            list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
+            expected_mass_pos)
+        self.assertEqual(list(obj.mass_pos_data[('92EB', '25')].keys()),
+                         expected_mass_pos)
+        self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
+
+    def test_gap(self):
+        expected_waveform = ['DS2-1', 'DS2-2', 'DS2-3']
+        with self.subTest("no gap_minimum set"):
+            args = {
+                'data_type': 'RT130',
+                'folder': reftek_gap_data,
+                'req_soh_chans': [],
+                'req_wf_chans': ['*'],
+                'rt130_waveform_data_req': True,
+                'on_unittest': True
+            }
+            obj = RT130(**args)
+            self.assertEqual(obj.found_data_streams, [2, 2])
+            self.assertEqual(obj.keys, [('98AD', '0')])
+            self.assertEqual(
+                list(obj.stream_header_by_key_chan[('98AD', '0')].keys()),
+                expected_waveform)
+
+            self.assertEqual(list(obj.waveform_data[('98AD', '0')].keys()),
+                             expected_waveform)
+            self.assertEqual(list(obj.log_data.keys()),
+                             ['TEXT', ('98AD', '0')])
+            self.assertEqual(obj.gaps[('98AD', '0')], [])
+
+        with self.subTest("has gap_minimum set"):
+            args = {
+                'data_type': 'RT130',
+                'folder': reftek_gap_data,
+                'req_soh_chans': [],
+                'req_wf_chans': ['*'],
+                'rt130_waveform_data_req': True,
+                'gap_minimum': 60,
+                'on_unittest': True
+            }
+            obj = RT130(**args)
+            self.assertEqual(obj.found_data_streams, [2, 2])
+            self.assertEqual(obj.keys, [('98AD', '0')])
+            self.assertEqual(
+                list(obj.stream_header_by_key_chan[('98AD', '0')].keys()),
+                expected_waveform)
+            self.assertEqual(list(obj.waveform_data[('98AD', '0')].keys()),
+                             expected_waveform)
+            self.assertEqual(list(obj.log_data.keys()),
+                             ['TEXT', ('98AD', '0')])
+            self.assertEqual(obj.gaps[('98AD', '0')],
+                             [[1648493999.64, 1648508400.64]])