Unit tests for parsing and reformatting of metadata fields

tests/test_lemi_metadata.py

+# -*- coding: utf-8 -*-
+
+"""Tests for `lemi_metadata` module."""
+
+import openpyxl
+import pickle
+import unittest
+
+from obspy import UTCDateTime
+from pathlib import Path
+
+from lemi2seed.lemi_data import LemiData
+from lemi2seed.lemi_metadata import LemiMetadata
+from lemi2seed.logging import setup_logger
+from lemi2seed.metadata_category import Elec, Mag, Aux
+
+OUTPUT_MSEED = Path(__file__).resolve().parent.joinpath('MSEED')
+OUTPUT_LOG = Path(__file__).resolve().parent.joinpath('LOG')
+OUTPUT_XML = Path(__file__).resolve().parent.joinpath('STATIONXML')
+TEST_DIR = Path(__file__).resolve().parent.joinpath('test_data')
+SCR_DIR = "lemi2seed.lemi_metadata"
+
+# Set up logging
+logger = setup_logger(SCR_DIR)
+
+
+class TestLemiMetadata(unittest.TestCase):
+    """Test suite for LemiMetadata class."""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST5"), OUTPUT_MSEED, OUTPUT_LOG)
+        lemi_data.prep_data()
+        self.data = lemi_data
+        self.path2md = TEST_DIR.joinpath("METADATA")
+        self.path2md_fail_1 = TEST_DIR.joinpath("METADATA_NO_FIELD_SHEET")
+        self.path2md_fail_2 = TEST_DIR.joinpath("METADATA_CORRUPTED")
+        self.file = self.path2md.joinpath('LEMI_Install_Sheet.xlsx')
+        self.md = LemiMetadata(self.path2md, OUTPUT_XML, self.data)
+        file_ = self.path2md.joinpath('metadata_fields.pkl')
+        with open(file_, 'rb') as fin:
+            self.md_fields_no_reformat = pickle.load(fin)
+        file_ = self.path2md.joinpath('metadata_fields_reformatted.pkl')
+        with open(file_, 'rb') as fin:
+            self.md_fields = pickle.load(fin)
+
+    def test_scan_path2md_field_sheet_provided(self):
+        """Test basic functionality of scan_path2md."""
+        md = LemiMetadata(self.path2md, OUTPUT_XML, self.data)
+        self.assertEqual(*md.filenames, self.file)
+
+    def test_scan_path2md_no_field_sheet_provided(self):
+        """Test basic functionality of scan_path2md."""
+        with self.assertLogs(logger, level='ERROR') as cmd:
+            LemiMetadata(self.path2md_fail_1, OUTPUT_XML, self.data)
+        msg = ("No field sheet found under the following path - {}. All "
+               "metadata will have to be provided using the GUI!"
+               .format(self.path2md_fail_1))
+        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
+
+    def test_get_md_field_pass(self):
+        """Test basic functionality of get_md_field."""
+        workbook = openpyxl.load_workbook(self.file, data_only=True)
+        sheet = workbook.active
+        self.assertEqual(LemiMetadata.get_md_field(sheet, 'B3'), 'EM')
+        workbook.close()
+
+    def test_get_md_field_fail(self):
+        """Test basic functionality of get_md_field."""
+        workbook = openpyxl.load_workbook(self.file, data_only=True)
+        sheet = workbook.active
+        with self.assertLogs(logger, level='ERROR'):
+            self.assertIsNone(LemiMetadata.get_md_field(sheet, 12))
+        workbook.close()
+
+    def test_identify_field_sheet_pass(self):
+        """Test basic functionality of identify_field_sheet."""
+        workbook = openpyxl.load_workbook(self.file, data_only=True)
+        sheet = workbook.active
+        sheet_type = self.md.identify_field_sheet(sheet, 'LEMI_Install_Sheet.xlsx')
+        self.assertEqual(sheet_type, 'install_sheet')
+        workbook.close()
+
+    def test_identify_field_sheet_fail(self):
+        """Test basic functionality of identify_field_sheet."""
+        md = LemiMetadata(self.path2md_fail_2, OUTPUT_XML, self.data)
+        file_ = self.path2md_fail_2.joinpath('LEMI_Install_Sheet_2.xlsx')
+        workbook = openpyxl.load_workbook(file_, data_only=True)
+        sheet = workbook.active
+        with self.assertLogs(logger, level='WARNING') as cmd:
+            md.identify_field_sheet(sheet, file_)
+        msg = ("The following file {} does not have the proper header. The "
+               "provided spread sheet templates were not used or their "
+               "layout was modified. Skipping file!".format(file_))
+        self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
+        workbook.close()
+
+    def test_reformat_run(self):
+        """Test basic functionality of reformat_run."""
+        run_fields = self.md_fields_no_reformat['Run']
+
+        file_ = self.path2md.joinpath('run_fields_reformatted.pkl')
+        with open(file_, 'rb') as fin:
+            reformatted = pickle.load(fin)
+        self.assertDictEqual(self.md.reformat_run(run_fields), reformatted)
+
+    def test_reformat_elec(self):
+        """Test basic functionality of reformat_elec."""
+        elec_fields = self.md_fields_no_reformat['Elec']
+        file_ = self.path2md.joinpath('electric_fields_reformatted.pkl')
+        with open(file_, 'rb') as fin:
+            reformatted = pickle.load(fin)
+        self.assertDictEqual(self.md.reformat_elec(elec_fields), reformatted)
+
+    def test_reformat_mag(self):
+        """Test basic functionality of reformat_mag."""
+        mag_fields = self.md_fields_no_reformat['Mag']
+        file_ = self.path2md.joinpath('magnetic_fields_reformatted.pkl')
+        with open(file_, 'rb') as fin:
+            reformatted = pickle.load(fin)
+        self.assertDictEqual(self.md.reformat_mag(mag_fields), reformatted)
+
+    def test_reformat_md_dict(self):
+        """Test basic functionality of reformat_md_dict."""
+        self.assertDictEqual(self.md.reformat_md_dict(self.md_fields_no_reformat),
+                             self.md_fields)
+
+    def test_from_field_sheets_failed_to_open(self):
+        """Test basic functionality of from_field_sheets"""
+        self.md.filenames = [self.path2md_fail_2.joinpath('LEMI_Install_Sheet_1.xlsx')]
+        with self.assertLogs(logger, level='ERROR') as cmd:
+            self.md.from_field_sheets
+        msg = ("Failed to open field sheet: {} - Skipping file - Exception: "
+               "File is not a zip file".format(self.md.filenames[0]))
+        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
+
+    def test_from_field_sheets_duplicate_sheet_type(self):
+        """Test basic functionality of from_field_sheets"""
+        self.md.filenames = [self.file] * 2
+        with self.assertLogs(logger, level='WARNING') as cmd:
+            self.md.from_field_sheets
+        msg = ("Already parsed the install sheet - You may have more than one "
+               "install sheet - Skipping {}!".format(self.file))
+        self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
+
+    def test_from_field_sheets_expected_workflow(self):
+        """Test basic functionality of from_field_sheets"""
+        self.assertDictEqual(self.md.from_field_sheets, self.md_fields)
+
+    def test_init_run_md_props(self):
+        """Test basic functionality of init_run_md_props."""
+        time_period_starts = [UTCDateTime(2020, 9, 30, 21, 5),
+                              UTCDateTime(2020, 9, 30, 21, 12),
+                              UTCDateTime(2020, 9, 30, 21, 14),
+                              UTCDateTime(2020, 10, 1, 0, 0)]
+        time_period_ends = [UTCDateTime(2020, 9, 30, 21, 11, 1),
+                            UTCDateTime(2020, 9, 30, 21, 13, 45),
+                            UTCDateTime(2020, 9, 30, 21, 27, 59),
+                            UTCDateTime(2020, 10, 1, 0, 5, 59)]
+        self.md.init_run_md_props()
+        for ind, run in enumerate(self.md.run):
+            self.assertEqual(run.start, time_period_starts[ind])
+            self.assertEqual(run.end, time_period_ends[ind])
+            self.assertEqual(run.resource_id, 'mt.run.id:' + self.md.run_list[ind])
+
+    def test_init_cha_md_props_elec(self):
+        """Test basic functionality of init_cha_md_props."""
+        cha = self.md.init_cha_md_props('elec', 'a')
+        self.assertIsInstance(cha, Elec)
+        self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3))
+        self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3))
+        self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3))
+        self.assertEqual(cha.run_id, 'a')
+
+    def test_init_cha_md_props_mag(self):
+        """Test basic functionality of init_cha_md_props."""
+        cha = self.md.init_cha_md_props('mag', 'a')
+        self.assertIsInstance(cha, Mag)
+        self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3))
+        self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3))
+        self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3))
+        self.assertEqual(cha.run_id, 'a')
+
+    def test_init_cha_md_props_aux(self):
+        """Test basic functionality of init_cha_md_props."""
+        cha = self.md.init_cha_md_props('aux', 'a')
+        self.assertIsInstance(cha, Aux)
+        self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3))
+        self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3))
+        self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3))
+        self.assertEqual(cha.run_id, 'a')
+
+    def test_dc2dict(self):
+        """Test basic functionality of dc2dict."""
+        cha = self.md.init_cha_md_props('elec', 'a')
+        self.assertDictEqual(self.md.dc2dict(cha),
+                             {'elev': 2201.725,
+                              'lat': 34.048,
+                              'lon': -107.128,
+                              'inst_manufacturer': None,
+                              'inst_model': None,
+                              'inst_specs': None,
+                              'inst_type': None,
+                              'meas_azimuth': None,
+                              'meas_tilt': None,
+                              'sample_rate': 1.0,
+                              'cha_num': None,
+                              'contact_resistance_end': None,
+                              'contact_resistance_start': None,
+                              'dc_end': None,
+                              'dc_start': None,
+                              'dipole_len': None,
+                              'neg_elec_dir': None,
+                              'neg_elec_sn': None,
+                              'pos_elec_dir': None,
+                              'pos_elec_sn': None})
+
+    def test_for_gui_md_not_populated(self):
+        """Test basic functionality of for_gui."""
+        file_ = self.path2md.joinpath('gui_metadata_not_populated.pkl')
+        with open(file_, 'rb') as fin:
+            expected = pickle.load(fin)
+        self.assertDictEqual(self.md.for_gui, expected)
+
+    def tearDown(self):
+        """Tear down test fixtures"""
+        file_ = self.path2md.joinpath('lemi_metadata.pkl')
+        if file_.is_file():
+            file_.unlink()