diff --git a/tests/test_model/test_handling_data.py b/tests/test_model/test_handling_data.py index 288f3d8cb93bf81cacd32605f0b7242960108cca..14ecbfaafb2085104bcddb9e66f11632f618fb42 100644 --- a/tests/test_model/test_handling_data.py +++ b/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()