From 5e074912b03934b9ba27473115e12f77072cbc09 Mon Sep 17 00:00:00 2001
From: ldam <ldam@passcal.nmt.edu>
Date: Tue, 29 Aug 2023 09:12:19 -0600
Subject: [PATCH] unittest for reftek
---
tests/model/reftek_data/__init__.py | 0
tests/model/reftek_data/test_reftek.py | 150 +++++++++++++++++++++++++
2 files changed, 150 insertions(+)
create mode 100644 tests/model/reftek_data/__init__.py
create mode 100644 tests/model/reftek_data/test_reftek.py
diff --git a/tests/model/reftek_data/__init__.py b/tests/model/reftek_data/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/model/reftek_data/test_reftek.py b/tests/model/reftek_data/test_reftek.py
new file mode 100644
index 000000000..8bb2a7174
--- /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]])
--
GitLab