Add test for readMPTrace

tests/test_model/test_handling_data.py

@@ -2,14 +2,17 @@ from pathlib import Path
 from math import isclose
 
 from unittest import TestCase
+from unittest.mock import patch
 
 from obspy.core import Stream, read as read_ms
 import numpy as np
 
 from sohstationviewer.model.handling_data import (
     readSOHMSeed,
-    readSOHTrace
+    readSOHTrace,
+    readMPTrace,
 )
+from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130
 
 TEST_DATA_DIR = Path(__file__).parent.parent.joinpath('test_data')
 
@@ -17,26 +20,24 @@ TEST_DATA_DIR = Path(__file__).parent.parent.joinpath('test_data')
 class TestHandlingData(TestCase):
     @classmethod
     def setUpClass(cls) -> None:
+        rt130_dir = TEST_DATA_DIR.joinpath(
+            'RT130-sample/2017149.92EB/2017150/92EB/9')
+        cls.rt130_file = rt130_dir.joinpath('054910000_013EE8A0')
+        rt130 = Reftek130.from_file(cls.rt130_file)
+        cls.rt130_stream = Reftek130.to_stream(rt130)
+        cls.rt130_trace = cls.rt130_stream[0]
+
         q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
-        # This file is chosen because it has time info.
         cls.q330_file = q330_dir.joinpath('AX08.XA..VKI.2021.186')
         cls.mseed_stream = read_ms(cls.q330_file)
-        cls.trace = cls.mseed_stream[0]
-        cls.processed_trace = {
-            'chanID': '',
-            'samplerate': '',
-            'startTmEpoch': '',
-            'endTmEpoch': '',
-            'times': '',
-            'data': '',
-        }
+        cls.mseed_trace = cls.mseed_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.trace)
+        processed_trace = readSOHTrace(self.mseed_trace)
         with self.subTest('test_processed_trace_have_all_needed_info'):
             expected_key_list = [
                 'chanID',
@@ -47,20 +48,34 @@ class TestHandlingData(TestCase):
                 'data'
             ]
             self.assertTrue(
-                all(key in processed_trace for key in expected_key_list)
+                all(key in processed_trace for key in expected_key_list),
+                msg='Processed trace is missing some fields.'
             )
 
     def test_read_soh_trace_times_calculated_correctly(self):
-        processed_trace = readSOHTrace(self.trace)
+        processed_trace = readSOHTrace(self.mseed_trace)
         if isclose(processed_trace['startTmEpoch'], 0, abs_tol=0.0001):
             self.assertAlmostEqual(processed_trace['times'][0], 0)
         else:
             self.assertNotAlmostEqual(processed_trace['times'][0], 0)
 
-    # @skip
-    # def test_read_mptrace(self):
-    #     self.fail()
-    #
+    @patch('sohstationviewer.model.handling_data.readSOHTrace')
+    def test_read_mp_trace(self, mock_read_soh_trace):
+        mock_read_soh_trace.return_value = {
+            # Contain five cases:
+            # + Small positive
+            # + Large positive
+            # + Small negative
+            # + Large negative
+            # + Zero
+            'data': np.array([1, 27272, -2, -23526, 0])
+        }
+        expected = np.array([0, 8.3, 0, -7.2, 0])
+        processed_trace = readMPTrace(self.rt130_trace)
+        self.assertTrue(
+            np.array_equal(processed_trace['data'], expected)
+        )
+
     # @skip
     # def test_read_waveform_trace(self):
     #     self.fail()