Unit tests for data conversion module

tests/test_data_conversion.py

+# -*- coding: utf-8 -*-
+
+"""Tests for `lemi_conversion` module."""
+
+import shutil
+import unittest
+
+from obspy import UTCDateTime
+from obspy.core import read
+from pathlib import Path
+
+from lemi2seed.data_conversion import (update_stats, init_stats,
+                                       write_daylong_mseed, write_log_file)
+from lemi2seed.lemi_data import LemiData, CHA_NAMING_CONV
+
+OUTPUT_MSEED = Path(__file__).resolve().parent.joinpath('MSEED')
+OUTPUT_LOG = Path(__file__).resolve().parent.joinpath('LOG')
+TEST_DIR = Path(__file__).resolve().parent.joinpath('test_data')
+
+
+class TestWriteLogFile(unittest.TestCase):
+    """Test suite for write_log_file function."""
+
+    def setUp(self):
+        lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST5"), OUTPUT_MSEED, OUTPUT_LOG)
+        lemi_data.prep_data()
+        user_inputs = {'net': 'EM', 'sta': 'KELLY',
+                       'e_chas': ['E1', 'E2', 'E3', 'E4', 'Hx', 'Hy', 'Hz'],
+                       'e_info': {'E1': 'Ex', 'E2': 'Ey', 'E3': 'Ex', 'E4': 'Ey'}}
+        lemi_data.update_data(qc_inputs=user_inputs)
+        self.data = lemi_data
+        self.log_filename = OUTPUT_LOG.joinpath(lemi_data.stats['sta'],
+                                                'LEMI424_2020274_205400_2020275_112705.log')
+
+    def test_write_log_file_exist(self):
+        """Test basic functionality of write_log_file."""
+        write_log_file(self.data)
+        self.assertTrue(self.log_filename.is_file())
+
+    def test_write_log_file_content(self):
+        """Test basic functionality of write_log_file."""
+        path_test_log = TEST_DIR.joinpath('LOGS')
+        test_log_file = path_test_log.joinpath('KELLY', self.log_filename.name)
+        with open(test_log_file, 'r') as fin:
+            test_def_lines = fin.readlines()
+        write_log_file(self.data)
+        with open(self.log_filename, 'r') as fin:
+            def_lines = fin.readlines()
+        self.assertListEqual(def_lines[1:], test_def_lines[1:])
+
+    def tearDown(self):
+        """Tear down test fixtures"""
+        if self.data.output_log.is_dir():
+            shutil.rmtree(self.data.output_log)
+
+
+class TestWriteDaylongMseed(unittest.TestCase):
+    """Test suite for write_daylong_mseed and *_stats functions."""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST1"), OUTPUT_MSEED, OUTPUT_LOG)
+        lemi_data.prep_data()
+        user_inputs = {'net': 'EM', 'sta': 'KELLY',
+                       'e_chas': ['E1', 'E2', 'E3', 'E4', 'Hx', 'Hy', 'Hz'],
+                       'e_info': {'E1': 'Ex', 'E2': 'Ey', 'E3': 'Ex', 'E4': 'Ey'}}
+        lemi_data.update_data(qc_inputs=user_inputs)
+        self.data = lemi_data
+
+    def test_init_stats(self):
+        """Test basic functionality of init_stats."""
+        stats = init_stats(self.data)
+        self.assertEqual(stats.network, 'EM')
+        self.assertEqual(stats.station, 'KELLY')
+        self.assertEqual(stats.sampling_rate, 1)
+
+    def test_update_stats(self):
+        """Test basic functionality of update_stats."""
+        stats = init_stats(self.data)
+        stats = update_stats(self.data.data_np[0], stats)
+        self.assertEqual(stats.channel, 'LQN')
+        self.assertEqual(stats.location, '00')
+        self.assertEqual(stats.starttime, UTCDateTime('2020-09-30T21:05:00.000000Z'))
+
+    def test_write_daylong_mseed(self):
+        """Test basic functionality of write_daylong_mseed."""
+        path_test_mseed = TEST_DIR.joinpath('MSEED')
+        write_daylong_mseed(self.data)
+        for cha_name in CHA_NAMING_CONV.values():
+            nbr_mseed = 2
+            if cha_name in ['LQN', 'LQE']:
+                nbr_mseed = 4
+            mseed_files = sorted(self.data.output_mseed.joinpath('KELLY').glob(f'*.{cha_name}.*'))
+            self.assertEqual(len(mseed_files), nbr_mseed)
+            for ind, mseed_file in enumerate(mseed_files):
+                st1 = read(str(mseed_file))
+                st2 = read(str(path_test_mseed.joinpath('KELLY', mseed_file.name)))
+                self.assertEqual(len(st1), 1)
+                self.assertListEqual(list(st1[0].data), list(st2[0].data))
+                self.assertEqual(st1[0].stats.starttime, st2[0].stats.starttime)
+                self.assertEqual(st1[0].stats.endtime, st2[0].stats.endtime)
+                self.assertEqual(st1[0].stats.network, st2[0].stats.network)
+                self.assertEqual(st1[0].stats.station, st2[0].stats.station)
+                self.assertEqual(st1[0].stats.location, st2[0].stats.location)
+                self.assertEqual(st1[0].stats.channel, st2[0].stats.channel)
+
+    def test_handle_existing_mseed(self):
+        """Test basic functionality of write_daylong_mseed."""
+        path_test_mseed = TEST_DIR.joinpath('MSEED')
+        write_daylong_mseed(self.data)
+        write_daylong_mseed(self.data)
+        for cha_name in CHA_NAMING_CONV.values():
+            mseed_files = self.data.output_mseed.joinpath('KELLY').glob(f'*.{cha_name}.*')
+            for mseed_file in mseed_files:
+                st1 = read(str(mseed_file))
+                st2 = read(str(path_test_mseed.joinpath('KELLY', mseed_file.name)))
+                for tr1, tr2 in zip(st1, st2):
+                    self.assertTrue(all(tr1.data == tr2.data))
+
+    def tearDown(self):
+        """Tear down test fixtures"""
+        if self.data.output_mseed.is_dir():
+            shutil.rmtree(self.data.output_mseed)