from tempfile import TemporaryDirectory, NamedTemporaryFile
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch
from contextlib import redirect_stdout
import io
from sohstationviewer.controller.processing import (
load_data,
read_mseed_channels,
detect_data_type,
get_data_type_from_file
)
from sohstationviewer.database.extract_data import get_signature_channels
from PySide2 import QtWidgets
from sohstationviewer.model.mseed.mseed import MSeed
from sohstationviewer.model.reftek.reftek import RT130
TEST_DATA_DIR = Path(__file__).resolve().parent.parent.joinpath('test_data')
rt130_dir = TEST_DATA_DIR.joinpath('RT130-sample/2017149.92EB/2017150')
q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08')
centaur_dir = TEST_DATA_DIR.joinpath('Centaur-sample/SOH')
pegasus_dir = TEST_DATA_DIR.joinpath('Pegasus-sample/Pegasus_SVC4/soh')
mix_traces_dir = TEST_DATA_DIR.joinpath('Q330_mixed_traces')
class TestLoadDataAndReadChannels(TestCase):
"""Test suite for load_data and read_mseed_channels."""
def setUp(self) -> None:
"""Set up test fixtures."""
patcher = patch.object(QtWidgets, 'QTextBrowser')
self.addCleanup(patcher.stop)
self.widget_stub = patcher.start()
# The actual value can be 'Q330', 'Centaur', or 'Pegasus'.
# The code that uses this string only cares that it is not 'RT130',
# though, so we are setting it to a stub value.
self.mseed_dtype = 'MSeed'
def test_load_data_rt130_good_dir(self):
"""
Test basic functionality of load_data - the given directory can be
loaded without issues. Test RT130.
"""
self.assertIsInstance(
load_data('RT130', self.widget_stub, [rt130_dir]),
RT130
)
def test_load_data_mseed_q330_good_data_dir(self):
"""
Test basic functionality of load_data - the given directory can be
loaded without issues. Test MSeed.
"""
self.assertIsInstance(
load_data(self.mseed_dtype, self.widget_stub, [q330_dir]),
MSeed
)
self.assertIsInstance(
load_data(self.mseed_dtype, self.widget_stub, [centaur_dir]),
MSeed
)
self.assertIsInstance(
load_data(self.mseed_dtype, self.widget_stub, [pegasus_dir]),
MSeed
)
def test_load_data_no_dir(self):
"""Test basic functionality of load_data - no directory was given."""
no_dir_given = []
self.assertIsNone(load_data('RT130', self.widget_stub, no_dir_given))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, no_dir_given))
def test_load_data_dir_does_not_exist(self):
"""
Test basic functionality of load_data - the given directory does not
exist.
"""
empty_name_dir = ['']
non_existent_dir = ['dir_that_does_not_exist']
self.assertIsNone(
load_data('RT130', self.widget_stub, empty_name_dir))
self.assertIsNone(
load_data('RT130', self.widget_stub, non_existent_dir))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, empty_name_dir))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, non_existent_dir))
def test_load_data_empty_dir(self):
"""
Test basic functionality of load_data - the given directory is empty.
"""
with TemporaryDirectory() as empty_dir:
self.assertIsNone(
load_data('RT130', self.widget_stub, [empty_dir]))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, [empty_dir]))
def test_load_data_empty_data_dir(self):
"""
Test basic functionality of load_data - the given directory
contains a data folder but no data file.
"""
with TemporaryDirectory() as outer_dir:
with TemporaryDirectory(dir=outer_dir) as data_dir:
self.assertIsNone(
load_data('RT130', self.widget_stub, [data_dir]))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, [outer_dir]))
def test_load_data_data_type_mismatch(self):
"""
Test basic functionality of load_data - the data type given does not
match the type of the data contained in the given directory.
"""
self.assertIsNone(
load_data('RT130', self.widget_stub, [q330_dir]))
self.assertIsNone(
load_data(self.mseed_dtype, self.widget_stub, [rt130_dir]))
def test_load_data_data_traceback_error(self):
"""
Test basic functionality of load_data - when there is an error
on loading data, the traceback info will be printed out
"""
f = io.StringIO()
with redirect_stdout(f):
self.assertIsNone(load_data('RT130', None, [q330_dir]))
output = f.getvalue()
self.assertIn(
f"Dir {q330_dir} "
f"can't be read due to error: Traceback",
output
)
with redirect_stdout(f):
self.assertIsNone(
load_data(self.mseed_dtype, None, [rt130_dir]))
output = f.getvalue()
self.assertIn(
f"Dir {rt130_dir} "
f"can't be read due to error: Traceback",
output
)
def test_read_channels_mseed_dir(self):
"""
Test basic functionality of load_data - the given directory contains
MSeed data.
"""
q330_soh_channels = sorted(['LOG', 'VKI'])
q330_mass_pos_channels = ['VM1']
q330_wf_channels = ['HHE', 'LHE']
q330_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [q330_dir], True)
self.assertListEqual(ret[0], q330_soh_channels)
self.assertListEqual(ret[1], q330_mass_pos_channels)
self.assertListEqual(ret[2], q330_wf_channels)
self.assertListEqual(ret[3], q330_spr_gt_1)
centaur_soh_channels = sorted(
['VDT', 'EX3', 'GEL', 'VEC', 'EX2', 'LCE', 'EX1', 'GLA', 'LCQ',
'GPL', 'GNS', 'GST', 'VCO', 'GAN', 'GLO', 'VPB', 'VEI'])
centaur_mass_pos_channels = sorted(['VM1', 'VM2', 'VM3'])
centaur_wf_channels = []
centaur_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [centaur_dir], True)
self.assertListEqual(ret[0], centaur_soh_channels)
self.assertListEqual(ret[1], centaur_mass_pos_channels)
self.assertListEqual(ret[2], centaur_wf_channels)
self.assertListEqual(ret[3], centaur_spr_gt_1)
pegasus_soh_channels = sorted(['VDT', 'VE1'])
pegasus_mass_pos_channels = sorted(['VM1'])
pegasus_wf_channels = []
pegasus_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [pegasus_dir], True)
self.assertListEqual(ret[0], pegasus_soh_channels)
self.assertListEqual(ret[1], pegasus_mass_pos_channels)
self.assertListEqual(ret[2], pegasus_wf_channels)
self.assertListEqual(ret[3], pegasus_spr_gt_1)
mix_traces_soh_channels = ['LOG']
mix_traces_mass_pos_channels = []
mix_traces_wf_channels = sorted(
['BH1', 'BH2', 'BH3', 'BH4', 'BH5', 'BH6',
'EL1', 'EL2', 'EL4', 'EL5', 'EL6', 'ELZ'])
mix_traces_spr_gt_1 = sorted(
['BS1', 'BS2', 'BS3', 'BS4', 'BS5', 'BS6',
'ES1', 'ES2', 'ES3', 'ES4', 'ES5', 'ES6',
'LS1', 'LS2', 'LS3', 'LS4', 'LS5', 'LS6',
'SS1', 'SS2', 'SS3', 'SS4', 'SS5', 'SS6'])
ret = read_mseed_channels(self.widget_stub, [mix_traces_dir], True)
self.assertListEqual(ret[0], mix_traces_soh_channels)
self.assertListEqual(ret[1], mix_traces_mass_pos_channels)
self.assertListEqual(ret[2], mix_traces_wf_channels)
self.assertListEqual(ret[3], mix_traces_spr_gt_1)
def test_read_channels_rt130_dir(self):
"""
Test basic functionality of load_data - the given directory contains
RT130 data.
"""
with self.assertRaises(Exception):
read_mseed_channels(self.widget_stub, [rt130_dir], True)
def test_read_mseed_channels_no_dir(self):
"""
Test basic functionality of read_mseed_channels - no directory was
given.
"""
no_dir = []
ret = read_mseed_channels(self.widget_stub, no_dir, True)
self.assertEqual(ret, ([], [], [], []))
def test_read_mseed_channels_dir_does_not_exist(self):
"""
Test basic functionality of read_mseed_channels - the given directory
does not exist.
"""
empty_name_dir = ['']
ret = read_mseed_channels(self.widget_stub, empty_name_dir, True)
self.assertEqual(ret, ([], [], [], []))
non_existent_dir = ['non_existent_dir']
ret = read_mseed_channels(self.widget_stub, non_existent_dir, True)
self.assertEqual(ret, ([], [], [], []))
def test_read_mseed_channels_empty_dir(self):
"""
Test basic functionality of read_mseed_channels - the given directory
is empty.
"""
with TemporaryDirectory() as empty_dir:
ret = read_mseed_channels(self.widget_stub, [empty_dir], True)
self.assertEqual(ret, ([], [], [], []))
def test_read_mseed_channels_empty_data_dir(self):
"""
Test basic functionality of read_mseed_channels - the given directory
contains a data folder but no data file.
"""
with TemporaryDirectory() as outer_dir:
with TemporaryDirectory(dir=outer_dir):
ret = read_mseed_channels(self.widget_stub, [outer_dir], True)
self.assertEqual(ret, ([], [], [], []))
class TestDetectDataType(TestCase):
"""
Test suite for detect_data_type and get_data_type_from_file functions.
"""
def setUp(self) -> None:
"""Set up text fixtures."""
widget_patcher = patch.object(QtWidgets, 'QTextBrowser')
self.addCleanup(widget_patcher.stop)
self.widget_stub = widget_patcher.start()
func_patcher = patch('sohstationviewer.controller.processing.'
'get_data_type_from_file')
self.addCleanup(func_patcher.stop)
self.mock_get_data_type_from_file = func_patcher.start()
self.dir1 = TemporaryDirectory()
self.dir2 = TemporaryDirectory()
self.file1 = NamedTemporaryFile(dir=self.dir1.name)
self.file2 = NamedTemporaryFile(dir=self.dir2.name)
def tearDown(self) -> None:
"""Teardown text fixtures."""
del self.file1, self.file2
self.dir1.cleanup()
self.dir2.cleanup()
def test_one_directory_not_unknown_data_type(self):
"""
Test basic functionality of detect_data_type - only one directory was
given and the data type it contains can be detected.
"""
expected_data_type = ('RT130', '_')
self.mock_get_data_type_from_file.return_value = expected_data_type
self.assertEqual(
detect_data_type(self.widget_stub, [self.dir1.name]),
expected_data_type[0]
)
def test_same_data_type_and_channel(self):
"""
Test basic functionality of detect_data_type - the given directories
contain the same data type and the data type was detected using the
same channel.
"""
expected_data_type = ('RT130', '_')
self.mock_get_data_type_from_file.return_value = expected_data_type
self.assertEqual(
detect_data_type(
self.widget_stub, [self.dir1.name, self.dir2.name]
),
expected_data_type[0]
)
def test_same_data_type_different_channel(self):
"""
Test basic functionality of detect_data_type - the given directories
contain the same data type but the data type was detected using
different channels.
"""
returned_data_types = [('Q330', 'OCF'), ('Q330', 'VEP')]
self.mock_get_data_type_from_file.side_effect = returned_data_types
self.assertEqual(
detect_data_type(
self.widget_stub, [self.dir1.name, self.dir2.name]
),
returned_data_types[0][0]
)
def test_different_data_types(self):
"""
Test basic functionality of detect_data_type - the given directories
contain different data types.
"""
returned_data_types = [('RT130', '_'), ('Q330', 'VEP')]
self.mock_get_data_type_from_file.side_effect = returned_data_types
self.assertIsNone(
detect_data_type(
self.widget_stub, [self.dir1.name, self.dir2.name]
)
)
def test_unknown_data_type(self):
"""
Test basic functionality of detect_data_type - can't detect any data
type.
"""
unknown_data_type = ('Unknown', '_')
self.mock_get_data_type_from_file.return_value = unknown_data_type
self.assertIsNone(detect_data_type(self.widget_stub, [self.dir1.name]))
class TestGetDataTypeFromFile(TestCase):
"""Test suite for get_data_type_from_file"""
def test_rt130_data(self):
"""
Test basic functionality of get_data_type_from_file - given file
contains RT130 data.
"""
rt130_file = Path(rt130_dir).joinpath(
'92EB/0/000000000_00000000')
expected_data_type = ('RT130', '_')
self.assertTupleEqual(
get_data_type_from_file(rt130_file, get_signature_channels()),
expected_data_type
)
def test_cannot_detect_data_type(self):
"""
Test basic functionality of get_data_type_from_file - cannot detect
data type contained in given file.
"""
test_file = NamedTemporaryFile()
self.assertIsNone(
get_data_type_from_file(test_file.name, get_signature_channels()))
def test_mseed_data(self):
"""
Test basic functionality of get_data_type_from_file - given file
contains MSeed data.
"""
q330_file = q330_dir.joinpath('AX08.XA..VKI.2021.186')
centaur_file = centaur_dir.joinpath(
'XX.3734.SOH.centaur-3_3734..20180817_000000.miniseed.miniseed')
pegasus_file = pegasus_dir.joinpath(
'2020/XX/KC01/VE1.D/XX.KC01..VE1.D.2020.129')
q330_data_type = ('Q330', 'VKI')
centaur_data_type = ('Centaur', 'GEL')
pegasus_data_type = ('Pegasus', 'VE1')
sig_chan = get_signature_channels()
self.assertTupleEqual(get_data_type_from_file(q330_file, sig_chan),
q330_data_type)
self.assertTupleEqual(get_data_type_from_file(centaur_file, sig_chan),
centaur_data_type)
self.assertTupleEqual(get_data_type_from_file(pegasus_file, sig_chan),
pegasus_data_type)
def test_file_does_not_exist(self):
"""
Test basic functionality of get_data_type_from_file - given file does
not exist.
"""
empty_name_file = ''
non_existent_file = 'non_existent_dir'
with self.assertRaises(FileNotFoundError):
get_data_type_from_file(empty_name_file, get_signature_channels())
with self.assertRaises(FileNotFoundError):
get_data_type_from_file(non_existent_file,
get_signature_channels())