diff --git a/tests/model/reftek_data/__init__.py b/tests/model/reftek_data/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/tests/model/reftek_data/test_reftek.py b/tests/model/reftek_data/test_reftek.py
new file mode 100644
index 0000000000000000000000000000000000000000..8bb2a71749d59d30833d4c98014308fff58060df
--- /dev/null
+++ b/tests/model/reftek_data/test_reftek.py
@@ -0,0 +1,150 @@
+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]])