Add tests for readWaveformMSeed

tests/test_model/test_handling_data.py

@@ -13,10 +13,13 @@ from sohstationviewer.model.handling_data import (
     readSOHTrace,
     readMPTrace,
     readWaveformTrace,
+    readWaveformMSeed,
 )
 from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130
 
 TEST_DATA_DIR = Path(__file__).parent.parent.joinpath('test_data')
+
+
 # tempfile.tempdir = './tempdir'
 
 
@@ -31,16 +34,19 @@ class TestHandlingData(TestCase):
         cls.rt130_trace = cls.rt130_stream[0]
 
         q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
-        cls.q330_file = q330_dir.joinpath('AX08.XA..VKI.2021.186')
-        cls.mseed_stream = read_ms(cls.q330_file)
-        cls.mseed_trace = cls.mseed_stream[0]
+        cls.q330_soh_file = q330_dir.joinpath('AX08.XA..VKI.2021.186')
+        cls.mseed_soh_stream = read_ms(cls.q330_soh_file)
+        cls.mseed_soh_trace = cls.mseed_soh_stream[0]
+        cls.q330_waveform_file = q330_dir.joinpath('AX08.XA..LHE.2021.186')
+        cls.mseed_waveform_stream = read_ms(cls.q330_waveform_file)
+        cls.mseed_waveform_trace = cls.mseed_waveform_stream[0]
 
     # @expectedFailure
     # def test_read_sohmseed(self):
     #     self.fail()
 
     def test_read_soh_trace_processed_trace_have_all_needed_info(self):
-        processed_trace = readSOHTrace(self.mseed_trace)
+        processed_trace = readSOHTrace(self.mseed_soh_trace)
         with self.subTest('test_processed_trace_have_all_needed_info'):
             expected_key_list = [
                 'chanID',
@@ -55,7 +61,7 @@ class TestHandlingData(TestCase):
             )
 
     def test_read_soh_trace_times_calculated_correctly(self):
-        processed_trace = readSOHTrace(self.mseed_trace)
+        processed_trace = readSOHTrace(self.mseed_soh_trace)
         if isclose(processed_trace['startTmEpoch'], 0, abs_tol=0.0001):
             self.assertAlmostEqual(processed_trace['times'][0], 0)
         else:
@@ -175,3 +181,97 @@ class TestHandlingData(TestCase):
     # @skip
     # def test_find_tpstm(self):
     #     self.fail()
+
+
+class TestReadWaveformMSeed(TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
+        cls.q330_waveform_file = q330_dir.joinpath('AX08.XA..LHE.2021.186')
+        cls.mseed_waveform_stream = read_ms(cls.q330_waveform_file)
+        cls.mseed_waveform_trace = cls.mseed_waveform_stream[0]
+
+    def setUp(self) -> None:
+        self.station_id = self.mseed_waveform_trace.stats['station']
+        self.channel_id = self.mseed_waveform_trace.stats['channel']
+        # This list is only ever written to, so we can keep it empty
+        self.traces_info = []
+
+        self.temp_dir = tempfile.TemporaryDirectory()
+
+        patcher = patch(
+            'sohstationviewer.model.handling_data.readWaveformTrace')
+        self.addCleanup(patcher.stop)
+        self.mock_readWaveformTrace = patcher.start()
+        self.mock_readWaveformTrace.return_value = {
+            'startTmEpoch': 0,
+            'endTmEpoch': 0,
+        }
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def test_all_traces_are_processed(self):
+        data_time = [12512512, 12215125121]
+        readWaveformMSeed(
+            self.q330_waveform_file, self.q330_waveform_file.name,
+            self.station_id, self.channel_id, self.traces_info, data_time,
+            self.temp_dir.name)
+        self.assertEqual(len(self.traces_info), len(self.mseed_waveform_stream))
+
+    def test_readWaveformTrace_called(self):
+        data_time = [12512512, 12215125121]
+        readWaveformMSeed(
+            self.q330_waveform_file, self.q330_waveform_file.name,
+            self.station_id, self.channel_id, self.traces_info, data_time,
+            self.temp_dir.name)
+        self.assertTrue(self.mock_readWaveformTrace.called)
+
+    def test_start_data_time_earlier_than_earliest_start_time(self):
+        self.mock_readWaveformTrace.return_value = {
+            'startTmEpoch': 51251,
+            'endTmEpoch': 2623623,
+        }
+        start_time = 0
+        data_time = [start_time, 1625532949]
+
+        readWaveformMSeed(
+            self.q330_waveform_file, self.q330_waveform_file.name,
+            self.station_id, self.channel_id, self.traces_info, data_time,
+            self.temp_dir.name)
+        self.assertEqual(data_time[0], start_time)
+
+    def test_end_data_time_later_than_latest_start_time(self):
+        # End time set to be the last second of 9999 so as to be later than the
+        # end time of any test data.
+        self.mock_readWaveformTrace.return_value = {
+            'startTmEpoch': 51251,
+            'endTmEpoch': 2623623,
+        }
+        end_time = 253402326000
+        data_time = [1625443222, end_time]
+
+        readWaveformMSeed(
+            self.q330_waveform_file, self.q330_waveform_file.name,
+            self.station_id, self.channel_id, self.traces_info, data_time,
+            self.temp_dir.name)
+        self.assertEqual(data_time[1], end_time)
+
+    def test_data_time_is_between_earliest_start_and_latest_end_time(self):
+        self.mock_readWaveformTrace.return_value = {
+            'startTmEpoch': 51251,
+            'endTmEpoch': 2623623,
+        }
+        start_time = 512579
+        end_time = 2623616
+        data_time = [start_time, end_time]
+
+        expected_updated_data_time = [51251, 2623623]
+        readWaveformMSeed(
+            self.q330_waveform_file, self.q330_waveform_file.name,
+            self.station_id, self.channel_id, self.traces_info, data_time,
+            self.temp_dir.name)
+        self.assertEqual(data_time, expected_updated_data_time)
+
+    # Test that dataTime is updated correctly
+