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

Add tests for readASCII

Change how incomplete tests fail themselves

Add explanation for a test of readASCII
parent 3f86cf24
No related branches found
No related tags found
1 merge request!27Draft: Add tests for functions in handling_data.py
from pathlib import Path
import tempfile
from math import isclose
from copy import copy as shallow_copy
from unittest import TestCase
from unittest.mock import patch
......@@ -14,7 +15,8 @@ from sohstationviewer.model.handling_data import (
readMPTrace,
readWaveformTrace,
readWaveformMSeed,
readWaveformReftek
readWaveformReftek,
readASCII,
)
from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130
......@@ -98,7 +100,8 @@ class TestHandlingData(TestCase):
traces_info = [dict() for _ in range(4)]
tmp_dir = tempfile.TemporaryDirectory()
processed_trace = readWaveformTrace(
self.rt130_soh_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 = [
......@@ -115,12 +118,6 @@ class TestHandlingData(TestCase):
self.assertTrue(mock_save_data_2_file.called)
# @skip
# def test_read_ascii(self):
# self.fail()
#
# @skip
# def test_read_text(self):
# self.fail()
......@@ -215,7 +212,8 @@ class TestReadWaveformMSeed(TestCase):
self.q330_waveform_file, self.q330_waveform_file.name,
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))
self.assertEqual(len(self.traces_info),
len(self.mseed_waveform_stream))
def test_readWaveformTrace_called(self):
readWaveformMSeed(
......@@ -376,3 +374,70 @@ class TestReadWaveformReftek(TestCase):
readWaveformReftek(self.rt130_waveform, self.key, self.read_data,
self.data_time, self.temp_dir.name)
self.assertEqual(self.data_time, expected_updated_data_time)
class TestReadASCII(TestCase):
@classmethod
def setUpClass(cls) -> None:
q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
cls.q330_log_file = q330_dir.joinpath('AX08.XA..LOG.2021.186')
cls.mseed_ascii_stream = read_ms(cls.q330_log_file)
cls.mseed_ascii_trace = cls.mseed_ascii_stream[0]
def track_info(self, msg, msg_type):
self.info_tracked.append((msg, msg_type))
def setUp(self) -> None:
self.station_id = self.mseed_ascii_trace.stats.station
self.channel_id = self.mseed_ascii_trace.stats.channel
self.log_data = {}
self.info_tracked = []
def test_station_and_channel_inserted_into_log_data(self):
readASCII(self.q330_log_file, None, self.station_id, self.channel_id,
self.mseed_ascii_trace, self.log_data, self.track_info)
self.assertTrue(self.station_id in self.log_data)
self.assertTrue(self.channel_id in self.log_data[self.station_id])
def test_trace_contains_log_data(self):
returned_file = readASCII(self.q330_log_file, None, self.station_id,
self.channel_id, self.mseed_ascii_trace,
self.log_data, self.track_info)
self.assertIsNone(returned_file)
log_string = self.log_data[self.station_id][self.channel_id][0]
log_lines = log_string.split('\n')
# Q330 uses \r\n as the new line character, so we need to remove the \r
# after splitting the log string based on \n
log_lines = [line.strip('\r') for line in log_lines]
# The first four elements of log_lines will be there whether anything
# is read from the file, so we know that something is read from the
# file when log_lines have more than four elements
self.assertGreater(len(log_lines), 4)
# Check that we are not reading in gibberish
self.assertEqual(
log_lines[4],
'Quanterra Packet Baler Model 14 Restart. Version 2.26'
)
def test_trace_does_not_contain_log_data(self):
raise NotImplementedError(self.test_trace_does_not_contain_log_data.__qualname__) # noqa
# We are only reassigning the data reference of the trace and are not
# modifying the stored data. As a result, we only need a shallow copy
# of the trace.
trace = shallow_copy(self.mseed_ascii_trace)
trace.data = np.array([])
returned_file = readASCII(self.q330_log_file, None, self.station_id,
self.channel_id, trace, self.log_data,
self.track_info)
self.assertIsNotNone(returned_file)
log_header = (
'\n\n**** STATE OF HEALTH: From:2021-07-05T03:37:40.120000Z '
'To:2021-07-05T03:37:40.120000Z\n'
)
log_info = self.mseed_ascii_trace.data.tobytes().decode()
log_string = log_header + log_info
print(self.log_data[self.station_id][self.channel_id][0])
self.assertEqual(log_string,
self.log_data[self.station_id][self.channel_id][0])
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