Skip to content
Snippets Groups Projects
Commit f9b61cfb authored by Kien Le's avatar Kien Le
Browse files

Move test local variable to setUp()

Move test local variable to setUp() in tests for readWaveformMSeed.
parent 37e48c7e
No related branches found
No related tags found
1 merge request!27Draft: Add tests for functions in handling_data.py
......@@ -14,6 +14,7 @@ from sohstationviewer.model.handling_data import (
readMPTrace,
readWaveformTrace,
readWaveformMSeed,
readWaveformReftek
)
from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130
......@@ -27,11 +28,15 @@ 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]
'RT130-sample/2017149.92EB/2017150/92EB')
cls.rt130_soh_file = rt130_dir.joinpath('9/054910000_013EE8A0')
rt130_soh = Reftek130.from_file(cls.rt130_soh_file)
cls.rt130_soh_stream = Reftek130.to_stream(rt130_soh)
cls.rt130_soh_trace = cls.rt130_soh_stream[0]
cls.rt130_waveform_file = rt130_dir.joinpath('9/054910000_013EE8A0')
rt130_waveform = Reftek130.from_file(cls.rt130_waveform_file)
cls.rt130_waveform_stream = Reftek130.to_stream(rt130_waveform)
cls.rt130_waveform_trace = cls.rt130_waveform_stream[0]
q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
cls.q330_soh_file = q330_dir.joinpath('AX08.XA..VKI.2021.186')
......@@ -79,21 +84,21 @@ class TestHandlingData(TestCase):
'data': np.array([1, 27272, -2, -23526, 0])
}
expected = np.array([0, 8.3, 0, -7.2, 0])
processed_trace = readMPTrace(self.rt130_trace)
processed_trace = readMPTrace(self.rt130_soh_trace)
self.assertTrue(
np.array_equal(processed_trace['data'], expected)
)
@patch('sohstationviewer.model.handling_data.saveData2File')
def test_read_waveform_trace(self, mock_save_data_2_file):
station_id = self.rt130_trace.stats['station']
channel_id = self.rt130_trace.stats['channel']
station_id = self.rt130_soh_trace.stats['station']
channel_id = self.rt130_soh_trace.stats['channel']
# The function itself only cares about the length of this list so we
# stub it out.
traces_info = [dict() for _ in range(4)]
tmp_dir = tempfile.TemporaryDirectory()
processed_trace = readWaveformTrace(
self.rt130_trace, station_id, channel_id, traces_info, tmp_dir.name
self.rt130_soh_trace, station_id, channel_id, traces_info, tmp_dir.name
)
expected_key_list = [
......@@ -110,14 +115,8 @@ class TestHandlingData(TestCase):
self.assertTrue(mock_save_data_2_file.called)
# @skip
# def test_read_waveform_mseed(self):
# self.fail()
#
# @skip
# def test_read_waveform_reftek(self):
# self.fail()
#
# @skip
# def test_read_ascii(self):
# self.fail()
......@@ -196,7 +195,7 @@ class TestReadWaveformMSeed(TestCase):
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.data_time = [12512512, 12215125121]
self.temp_dir = tempfile.TemporaryDirectory()
patcher = patch(
......@@ -212,18 +211,16 @@ class TestReadWaveformMSeed(TestCase):
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.station_id, self.channel_id, self.traces_info, self.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.station_id, self.channel_id, self.traces_info, self.data_time,
self.temp_dir.name)
self.assertTrue(self.mock_readWaveformTrace.called)
......@@ -233,13 +230,13 @@ class TestReadWaveformMSeed(TestCase):
'endTmEpoch': 2623623,
}
start_time = 0
data_time = [start_time, 1625532949]
self.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.station_id, self.channel_id, self.traces_info, self.data_time,
self.temp_dir.name)
self.assertEqual(data_time[0], start_time)
self.assertEqual(self.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
......@@ -249,13 +246,13 @@ class TestReadWaveformMSeed(TestCase):
'endTmEpoch': 2623623,
}
end_time = 253402326000
data_time = [1625443222, end_time]
self.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.station_id, self.channel_id, self.traces_info, self.data_time,
self.temp_dir.name)
self.assertEqual(data_time[1], end_time)
self.assertEqual(self.data_time[1], end_time)
def test_data_time_is_between_earliest_start_and_latest_end_time(self):
self.mock_readWaveformTrace.return_value = {
......@@ -264,14 +261,15 @@ class TestReadWaveformMSeed(TestCase):
}
start_time = 512579
end_time = 2623616
data_time = [start_time, end_time]
self.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.station_id, self.channel_id, self.traces_info, self.data_time,
self.temp_dir.name)
self.assertEqual(data_time, expected_updated_data_time)
self.assertEqual(self.data_time, expected_updated_data_time)
# Test that dataTime is updated correctly
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment