tests/test_model/test_reftek/test_header.py

+import unittest
+
+from sohstationviewer.model.reftek_data.reftek_reader.header import (
+    parse_rt130_time,
+    get_rt130_packet_header, NotRT130FileError,
+)
+
+
+class TestParseRT130Time(unittest.TestCase):
+    def test_time_bytes_parsed_correctly(self):
+        time_bytes = b'\x36\x01\x15\x13\x51\x35'
+        year = 15
+        result = parse_rt130_time(year, time_bytes)
+        self.assertEqual(result.julday, 360)
+        self.assertEqual(result.day, 26)
+        self.assertEqual(result.month, 12)
+        self.assertEqual(result.hour, 11)
+        self.assertEqual(result.minute, 51)
+        self.assertEqual(result.second, 35)
+        self.assertEqual(result.microsecond, 135000)
+        self.assertEqual(result.ns, 1451130695135000000)
+
+    def test_year_1900s(self):
+        time_bytes = b'\x36\x01\x15\x13\x51\x35'
+        year = 71
+        result = parse_rt130_time(year, time_bytes)
+        self.assertEqual(result.year, 1971)
+
+    def test_year_2000s(self):
+        time_bytes = b'\x36\x01\x15\x13\x51\x35'
+        year = 12
+        result = parse_rt130_time(year, time_bytes)
+        self.assertEqual(result.year, 2012)
+
+    def test_year_threshold(self):
+        with self.subTest('test_year_is_49'):
+            time_bytes = b'\x36\x01\x15\x13\x51\x35'
+            year = 49
+            result = parse_rt130_time(year, time_bytes)
+            self.assertEqual(result.year, 2049)
+        with self.subTest('test_year_is_50'):
+            time_bytes = b'\x36\x01\x15\x13\x51\x35'
+            year = 50
+            result = parse_rt130_time(year, time_bytes)
+            self.assertEqual(result.year, 1950)
+
+
+class TestGetRT130PacketHeader(unittest.TestCase):
+    def test_header_extracted_correctly(self):
+        header = b'DT\x12\x15\x98\xe1\x36\x01\x15\x13\x51\x35\x05\x12\x01\x11'
+        packet = header + b' ' * 1008
+        result = get_rt130_packet_header(packet)
+        self.assertEqual(result.packet_type, 'DT')
+        self.assertEqual(result.experiment_number, 12)
+        self.assertEqual(result.unit_id, '98E1')
+        self.assertEqual(result.time.ns, 1451130695135000000)
+        self.assertEqual(result.byte_count, 512)
+        self.assertEqual(result.packet_sequence, 111)
+
+    def test_packet_type_cannot_be_parsed(self):
+        packet_type = b'\x01\x02'
+        header = packet_type + b'\x11' * 14
+        packet = header + b' ' * 1008
+        with self.assertRaises(NotRT130FileError):
+            get_rt130_packet_header(packet)
+
+    def test_packet_type_is_not_valid(self):
+        packet_type = b'AB'
+        header = packet_type + b'\x11' * 14
+        packet = header + b' ' * 1008
+        with self.assertRaises(NotRT130FileError):
+            get_rt130_packet_header(packet)

tests/test_model/test_reftek/test_packet_readers.py

+import unittest
+from unittest.mock import patch
+
+from sohstationviewer.model.mseed_data.record_reader_helper import Unpacker
+from sohstationviewer.model.reftek_data.reftek_reader.packet import \
+    eh_et_payload_end_in_packet
+from sohstationviewer.model.reftek_data.reftek_reader.packet_readers import (
+    decode_uncompressed, decode_compressed, read_dt_packet, read_eh_et_packet,
+    read_soh_packet,
+)
+from sohstationviewer.model.reftek_data.reftek_reader.packets import \
+    SOHExtendedHeader
+
+unpacker = Unpacker('>')
+
+
+class TestDecodeFunctions(unittest.TestCase):
+    def setUp(self) -> None:
+        self.header = b' ' * 24
+
+    def test_decode_uncompressed_format_16(self):
+        data_format = '16'
+        with self.subTest('test_positive_number'):
+            first_data_point_byte = b'\x06\x19'
+            data_filler = b' ' * 998
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = 1561
+            self.assertEqual(actual, expected)
+        with self.subTest('test_negative_number'):
+            first_data_point_byte = b'\xf9\xe4'
+            data_filler = b' ' * 998
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = -1564
+            self.assertEqual(actual, expected)
+
+    def test_decode_uncompressed_format_32(self):
+        data_format = '32'
+        with self.subTest('test_positive_number'):
+            first_data_point_byte = b'\x03\xc5\x4e\x9a'
+            data_filler = b' ' * 996
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = 63262362
+            self.assertEqual(actual, expected)
+        with self.subTest('test_negative_number'):
+            first_data_point_byte = b'\xf6\xac\xba\x00'
+            data_filler = b' ' * 996
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = -156452352
+            self.assertEqual(actual, expected)
+
+    def test_decode_uncompressed_format_33(self):
+        data_format = '33'
+        with self.subTest('test_positive_number'):
+            first_data_point_byte = b'\x03\xc5\x4e\x9a'
+            data_filler = b' ' * 996
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = 63262362
+            self.assertEqual(actual, expected)
+        with self.subTest('test_negative_number'):
+            first_data_point_byte = b'\xf6\xac\xba\x00'
+            data_filler = b' ' * 996
+            packet = self.header + first_data_point_byte + data_filler
+            actual = decode_uncompressed(packet, data_format, unpacker)
+            expected = -156452352
+            self.assertEqual(actual, expected)
+
+    def test_decode_compressed(self):
+        data_format = 'C0'
+        filler = b' ' * 40
+        first_frame_code = b'\x00\x11\x11\x11'
+        start_data_point_byte = b'0000'
+        header = self.header + filler
+        bytes_before_data = header + first_frame_code + start_data_point_byte
+        with self.subTest('test_positive_number'):
+            end_point_byte = b'\x03\xc5\x4e\x9a'
+            data_filler = b' ' * 952
+            packet = bytes_before_data + end_point_byte + data_filler
+            actual = decode_compressed(packet, data_format, unpacker)
+            expected = 63262362
+            self.assertEqual(actual, expected)
+        with self.subTest('test_negative_number'):
+            end_point_byte = b'\xf6\xac\xba\x00'
+            data_filler = b' ' * 952
+            packet = bytes_before_data + end_point_byte + data_filler
+            actual = decode_compressed(packet, data_format, unpacker)
+            expected = -156452352
+            self.assertEqual(actual, expected)
+
+
+class TestReadDTPacket(unittest.TestCase):
+    def setUp(self) -> None:
+        self.header = b' ' * 16
+        # We only test if the correct method is used to extract the data point,
+        # so the data can be anything we want.
+        self.data = b' ' * 1000
+
+        uncompressed_patcher = patch(
+            'sohstationviewer.model.reftek_data.reftek_reader.packet_readers.'
+            'decode_uncompressed'
+        )
+        compressed_patcher = patch(
+            'sohstationviewer.model.reftek_data.reftek_reader.packet_readers.'
+            'decode_compressed'
+        )
+        self.mock_uncompressed = uncompressed_patcher.start()
+        self.mock_compressed = compressed_patcher.start()
+        self.addCleanup(uncompressed_patcher.stop)
+        self.addCleanup(compressed_patcher.stop)
+
+    def test_extended_header_is_extracted_correctly(self):
+        extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\xc0'
+        packet = self.header + extended_header_bytes + self.data
+        extended_header, _ = read_dt_packet(packet, unpacker)
+        self.assertEqual(extended_header.event_number, 111)
+        self.assertEqual(extended_header.data_stream_number, 1)
+        self.assertEqual(extended_header.channel_number, 2)
+        self.assertEqual(extended_header.number_of_samples, 500)
+        self.assertEqual(extended_header.flags, 0)
+        self.assertEqual(extended_header.data_format, 'C0')
+
+    def test_data_point_extracted_with_correct_method(self):
+        with self.subTest('test_uncompressed_packet'):
+            extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\x16'
+            packet = self.header + extended_header_bytes + self.data
+            read_dt_packet(packet, unpacker)
+            self.assertTrue(self.mock_uncompressed.called)
+            self.assertFalse(self.mock_compressed.called)
+
+        self.mock_uncompressed.reset_mock()
+        self.mock_compressed.reset_mock()
+
+        with self.subTest('test_compressed_packet'):
+            extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\xc0'
+            packet = self.header + extended_header_bytes + self.data
+            read_dt_packet(packet, unpacker)
+            self.assertTrue(self.mock_compressed.called)
+            self.assertFalse(self.mock_uncompressed.called)
+
+
+class TestReadEHETPacket(unittest.TestCase):
+    def setUp(self) -> None:
+        header = b' ' * 16
+        extended_header_bytes = b'\x01\x11\x01\x00\x00\x00\x00\xc0'
+        # We only care about the length of the payload (the content is dealt
+        # with somewhere else), and so it can contain dummy data.
+        payload = b' ' * 1000
+        self.packet = header + extended_header_bytes + payload
+
+    def test_extended_header_is_extracted_correctly(self):
+        extended_header, _ = read_eh_et_packet(self.packet, unpacker)
+        self.assertEqual(extended_header.event_number, 111)
+        self.assertEqual(extended_header.data_stream_number, 1)
+        self.assertEqual(extended_header.channel_number, 0)
+        self.assertEqual(extended_header.number_of_samples, 0)
+        self.assertEqual(extended_header.flags, 0)
+        self.assertEqual(extended_header.data_format, 'C0')
+
+    def test_payload_extracted_correctly(self):
+        _, payload = read_eh_et_packet(self.packet, unpacker)
+        self.assertEqual(len(payload), eh_et_payload_end_in_packet - 24)
+
+
+class TestReadSOHPacket(unittest.TestCase):
+    """
+    Test suite for packet_readers.read_soh_packet. We only test that the
+    function has the correct interface, seeing as the intended purpose of this
+    method is to be compatible with packet_readers.read_dt_packet and
+    packet_readers.read_eh_et_packet interface-wise.
+    """
+    def test_correct_interface(self):
+        packet = b' ' * 1024
+        extended_header, payload = read_soh_packet(packet, unpacker)
+        self.assertIsInstance(extended_header, SOHExtendedHeader)
+        self.assertIsInstance(payload, bytes)
+
+    def test_payload_has_correct_length(self):
+        packet = b' ' * 1024
+        extended_header, payload = read_soh_packet(packet, unpacker)
+        self.assertEqual(len(payload), 1000)

tests/test_model/test_reftek/test_reftek_helper.py

+import os
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from obspy.io.reftek.packet import PACKET_FINAL_DTYPE
+
+from sohstationviewer.model.mseed_data.record_reader_helper import Unpacker
+from sohstationviewer.model.reftek_data.reftek_reader.header import \
+    NotRT130FileError
+from sohstationviewer.model.reftek_data.reftek_reader.packet_readers import (
+    read_eh_et_packet, read_dt_packet, read_soh_packet,
+)
+from sohstationviewer.model.reftek_data.reftek_reader.packets import (
+    SOHPacket,
+    EHETPacket, DTPacket,
+)
+from sohstationviewer.model.reftek_data.reftek_reader.reftek_reader_helper \
+    import (read_rt130_file, convert_packet_to_obspy_format)
+
+unpacker = Unpacker('>')
+
+
+class TestReadRT130File(unittest.TestCase):
+    def setUp(self) -> None:
+        self.TEST_DATA_DIR = Path(os.getcwd()).joinpath('tests/test_data')
+        self.rt130_dir = self.TEST_DATA_DIR.joinpath(
+            'RT130-sample/2017149.92EB/2017150/92EB'
+        )
+
+        eh_et_patcher = patch(
+            'sohstationviewer.model.reftek_data.reftek_reader.'
+            'reftek_reader_helper.'
+            'read_eh_et_packet',
+            wraps=read_eh_et_packet
+        )
+        self.mock_read_eh_et = eh_et_patcher.start()
+        self.addCleanup(eh_et_patcher.stop)
+
+        dt_patcher = patch(
+            'sohstationviewer.model.reftek_data.reftek_reader.'
+            'reftek_reader_helper.'
+            'read_dt_packet',
+            wraps=read_dt_packet
+        )
+        self.mock_read_dt = dt_patcher.start()
+        self.addCleanup(dt_patcher.stop)
+
+        soh_patcher = patch(
+            'sohstationviewer.model.reftek_data.reftek_reader.'
+            'reftek_reader_helper.'
+            'read_soh_packet',
+            wraps=read_soh_packet
+        )
+        self.mock_read_soh = soh_patcher.start()
+        self.addCleanup(soh_patcher.stop)
+
+    def test_rt130_soh_file(self):
+        file = self.rt130_dir.joinpath('0/000000000_00000000')
+        packets = read_rt130_file(file, unpacker)
+        self.assertTrue(
+            all(isinstance(packet, SOHPacket) for packet in packets)
+        )
+        self.assertTrue(self.mock_read_soh.called)
+        self.assertFalse(self.mock_read_dt.called)
+        self.assertFalse(self.mock_read_eh_et.called)
+
+    def test_rt130_raw_data_file(self):
+        file = self.rt130_dir.joinpath('1/000000015_0036EE80')
+        packets = read_rt130_file(file, unpacker)
+        self.assertTrue(all(
+            isinstance(packet, EHETPacket) or isinstance(packet, DTPacket)
+            for packet in packets)
+        )
+        self.assertFalse(self.mock_read_soh.called)
+        self.assertTrue(self.mock_read_dt.called)
+        self.assertTrue(self.mock_read_eh_et.called)
+
+    def test_non_rt130_file(self):
+        with self.subTest('test_file_exist'):
+            file = self.TEST_DATA_DIR.joinpath(
+                'Q330-sample/day_vols_AX08/AX08.XA..HHE.2021.186'
+            )
+            with self.assertRaises(NotRT130FileError):
+                read_rt130_file(file, unpacker)
+
+        with self.subTest('test_file_does_not_exist'):
+            file = ''
+            with self.assertRaises(FileNotFoundError):
+                read_rt130_file(file, unpacker)
+
+
+class TestConvertPacketToObspyFormat(unittest.TestCase):
+    def setUp(self) -> None:
+        TEST_DATA_DIR = Path(os.getcwd()).joinpath('tests/test_data')
+        rt130_dir = TEST_DATA_DIR.joinpath(
+            'RT130-sample/2017149.92EB/2017150/92EB'
+        )
+        file = rt130_dir.joinpath('1/000000015_0036EE80')
+        self.packet = read_rt130_file(file, unpacker)[0]
+
+    def test_all_needed_fields_are_available(self):
+        converted_packet = convert_packet_to_obspy_format(
+            self.packet, unpacker
+        )
+
+        self.assertEqual(len(converted_packet), len(PACKET_FINAL_DTYPE))

tests/test_view/test_util_functions.py

@@ -9,7 +9,8 @@ from sohstationviewer.view.util.functions import (
     get_soh_messages_for_view, log_str, is_doc_file,
     create_search_results_file, create_table_of_content_file,
     check_chan_wildcards_format, check_masspos, get_total_miny_maxy,
-    extract_netcodes, get_index_from_time
+    extract_netcodes, get_index_from_time, remove_not_found_chans,
+    replace_actual_question_chans
 )
 
 from sohstationviewer.view.util.enums import LogType
@@ -501,3 +502,36 @@ class TestGetIndexFromTime(TestCase):
                 self.plotting_data['CH2'], 3, 4)
             self.assertEqual(list_idx, 1)
             self.assertEqual(section_idx, 0)
+
+
+class RemoveNotFoundChansClass(TestCase):
+    def test_remove_not_found_chans(self):
+        chan_order = ['A', 'B', 'C', 'D']
+        actual_chans = ['C', 'D', 'E', 'F']
+        processing_log = []
+        expected_new_chan_order = ['C', 'D']
+        expected_processing_log = [
+            ("No data found for the following channels: A, B",
+             LogType.WARNING)]
+
+        ret = remove_not_found_chans(chan_order, actual_chans, processing_log)
+        self.assertListEqual(ret, expected_new_chan_order)
+        self.assertEqual(processing_log, expected_processing_log)
+
+
+class ReplaceActualQuestChans(TestCase):
+    def test_question_chans_in_actual_chans(self):
+        chan_order = ['A', 'B', 'C?', 'D']
+        actual_chans = ['C1', 'C3', 'C2', 'D', 'E', 'F']
+        expected_new_chan_order = ['A', 'B', 'C1', 'C2', 'C3', 'D']
+
+        ret = replace_actual_question_chans(chan_order, actual_chans)
+        self.assertListEqual(ret, expected_new_chan_order)
+
+    def test_question_chans_not_in_actual_chans(self):
+        chan_order = ['A?', 'B', 'C', 'D']
+        actual_chans = ['C', 'D', 'E', 'F']
+        expected_new_chan_order = ['A?', 'B', 'C', 'D']
+
+        ret = replace_actual_question_chans(chan_order, actual_chans)
+        self.assertListEqual(ret, expected_new_chan_order)