test_metadata_category.py

# -*- coding: utf-8 -*-

"""Tests for `metadata_category` module."""

import unittest

from obspy import UTCDateTime
from pathlib import Path

from lemi2seed.lemi_data import LemiData, VALID_COMPS
from lemi2seed.metadata_category import (BaseNet, Net, BaseSta, Sta, Run, Elec,
                                         Mag, Aux)
from lemi2seed.logging import setup_logger
from lemi2seed.utils import MSG_E_CHA

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')

SCR_DIR = "lemi2seed.metadata_category"

# Set up logging
logger = setup_logger(SCR_DIR)


class TestBaseNet(unittest.TestCase):
    """Test suite for BaseNet data class."""

    def setUp(self):
        """Set up test fixtures"""
        lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST1"), OUTPUT_MSEED, OUTPUT_LOG)
        lemi_data.prep_data()
        self.data_stats = lemi_data.stats

    def test_validate_start_not_utc(self):
        """Test basic functionality of validate_start."""
        data_input = self.data_stats['start']
        md_input = 2021.167
        bs = BaseNet()
        self.assertFalse(bs.validate_start(md_input, data_input))

    def test_validate_start_greater_than_acq_start(self):
        """Test basic functionality of validate_start."""
        data_input = self.data_stats['start']
        md_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
        bs = BaseNet()
        self.assertFalse(bs.validate_start(md_input, data_input))

    def test_validate_start_valid(self):
        """Test basic functionality of validate_start."""
        data_input = self.data_stats['start']
        md_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
        bs = BaseNet()
        self.assertTrue(bs.validate_start(md_input, data_input))

    def test_validate_end_not_utc(self):
        """Test basic functionality of validate_end."""
        data_input = self.data_stats['end']
        md_input = 2021.167
        bs = BaseNet()
        self.assertFalse(bs.validate_end(md_input, data_input))

    def test_validate_end_lower_than_acq_end(self):
        """Test basic functionality of validate_end."""
        data_input = self.data_stats['end']
        md_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
        bs = BaseNet()
        self.assertFalse(bs.validate_end(md_input, data_input))

    def test_validate_end_valid(self):
        """Test basic functionality of validate_end."""
        data_input = self.data_stats['end']
        md_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
        bs = BaseNet()
        self.assertTrue(bs.validate_end(md_input, data_input))


class TestNet(unittest.TestCase):
    """Test suite for Net data class."""

    def test_validate_archive_net_undefined(self):
        """Test basic functionality of validate_archive_net."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_archive_net(None)
        msg = "The network code should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_net_erroneous_type(self):
        """Test basic functionality of validate_archive_net."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_archive_net(12)
        msg = "The network code should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_net_invalid(self):
        """Test basic functionality of validate_archive_net."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_archive_net('EMX')
        msg = "The network code should be two alphanumeric character long."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_net_valid(self):
        """Test basic functionality of validate_archive_net."""
        self.assertTrue(Net.validate_archive_net('EM'))

    def test_validate_citation_dataset_doi_undefined(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        doi = None
        self.assertTrue(Net.validate_citation_dataset_doi(doi))

    def test_validate_citation_dataset_doi_erroneous_type(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_citation_dataset_doi(10.7914)
        msg = "The DOI number(s) should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_citation_dataset_doi_invalid_doi(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_citation_dataset_doi('10.7914/SN/EM')
        msg = ("Invalid DOI(s). The DOI number(s) provided by the archive "
               "should be strings formatted as follows: 'scheme: path'.")
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_citation_dataset_doi_invalid_dois(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        dois = '10.7914/SN/EM, DOI:10.3421/SN/EG'
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_citation_dataset_doi(dois)
        msg = ("Invalid DOI(s). The DOI number(s) provided by the archive "
               "should be strings formatted as follows: 'scheme: path'.")
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_citation_dataset_doi_valid_doi(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        doi = 'DOI:10.7914/SN/EM'
        self.assertTrue(Net.validate_citation_dataset_doi(doi))

    def test_validate_citation_dataset_doi_valid_dois(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        dois = 'DOI:10.7914/SN/EM, DOI:10.3421/SN/EG'
        self.assertTrue(Net.validate_citation_dataset_doi(dois))

    def test_validate_project_lead_email_undefined(self):
        """Test basic functionality of validate_project_lead_email."""
        self.assertTrue(Net.validate_project_lead_email(None))

    def test_validate_project_lead_email_erroneous_type(self):
        """Test basic functionality of validate_project_lead_email."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Net.validate_project_lead_email(12)
        msg = "The project lead email(s) should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_project_lead_email_invalid_email(self):
        """Test basic functionality of validate_project_lead_email."""
        email = 'mpasscal.edu'
        self.assertFalse(Net.validate_project_lead_email(email))

    def test_validate_project_lead_email_invalid_emails(self):
        """Test basic functionality of validate_project_lead_email."""
        emails = 'mpasscal.edu, d@passcal.edu'
        self.assertFalse(Net.validate_project_lead_email(emails))

    def test_validate_project_lead_email_valid_email(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        email = 'm@passcal.edu'
        self.assertTrue(Net.validate_project_lead_email(email))

    def test_validate_project_lead_email_valid_emails(self):
        """Test basic functionality of validate_citation_dataset_doi."""
        emails = 'm@passcal.edu, d@passcal.edu'
        self.assertTrue(Net.validate_project_lead_email(emails))


class TestBaseSta(unittest.TestCase):
    """Test suite for BaseSta data class."""

    def setUp(self):
        """Set up test fixtures"""
        data_input = 2201.77
        min_elev, max_elev = [abs(data_input) * x for x in [0.99, 1.01]]
        param = ("station elevation", -500, 8500, min_elev, max_elev, data_input, 'm', '')
        self.param = param

    def test_validate_geos_undefined(self):
        """Test basic functionality of validate_geos."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            BaseSta.validate_geos(self.param, None)
        msg = "Station elevation should be a float. "
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_geos_erroneous_type(self):
        """Test basic functionality of validate_geos."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            BaseSta.validate_geos(self.param, 'a')
        msg = "Station elevation should be a float. "
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_geos_not_in_range(self):
        """Test basic functionality of validate_geos."""
        geo, min_range, max_range, _, _, _, units, _ = self.param
        with self.assertLogs(logger, level='ERROR') as cmd:
            BaseSta.validate_geos(self.param, 9000)
        msg = ("Unexpected {0}! The {0} should be between {1}{3} and {2}{3}."
               .format(geo, min_range, max_range, units))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_geos_do_not_match_logger_recorded_metadata(self):
        """Test basic functionality of validate_geos."""
        geo, _, _, _, _, val, units, _ = self.param
        with self.assertLogs(logger, level='ERROR') as cmd:
            BaseSta.validate_geos(self.param, 2250)
        msg = ("Unexpected {0}! Provided {0} should roughly match the {0} "
               "recorded by the on-site GPS ({1:.3f}{2}).".format(geo, val, units))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])


class TestSta(unittest.TestCase):
    """Test suite for Sta data class."""

    def test_validate_archive_id_undefined(self):
        """Test basic functionality of validate_archive_id."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Sta.validate_archive_id(None)
        msg = "The station name should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_id_erroneous_type(self):
        """Test basic functionality of validate_archive_id."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Sta.validate_archive_id(12)
        msg = "The station name should be a string."
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_id_invalid(self):
        """Test basic functionality of validate_archive_id."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            Sta.validate_archive_id('KELLYA')
        msg = ("The station name should be between three and five "
               "alphanumeric character long.")
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_archive_id_valid(self):
        """Test basic functionality of validate_archive_id."""
        self.assertTrue(Sta.validate_archive_id('KELLY'))

    def test_validate_submitter_email_undefined(self):
        """Test basic functionality of validate_submitter_email."""
        self.assertTrue(Sta.validate_submitter_email(None))


class TestRun(unittest.TestCase):
    """Test suite for Run data class."""

    def setUp(self):
        """Set up test fixtures"""
        self.run = Run(resource_id='mt.run.id:a')
        lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST1"), OUTPUT_MSEED, OUTPUT_LOG)
        lemi_data.prep_data()
        self.data_stats = lemi_data.stats

    def test_validate_comps_rec_undefined(self):
        """Test basic functionality of validate_comps_rec."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.run.validate_comps_rec(None)
        msg = ("The list of components recorded for run '{}' should be a "
               "string.".format(self.run.run_id))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_comps_rec_erroneous_type(self):
        """Test basic functionality of validate_comps_rec."""
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.run.validate_comps_rec(12)
        msg = ("The list of components recorded for run '{}' should be a "
               "string.".format(self.run.run_id))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_comps_rec_invalid(self):
        """Test basic functionality of validate_comps_recorded."""
        comps = 'E1, E2, Hx, Hy, Hn'
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.run.validate_comps_rec(comps)
        msg = ("Some of the listed components for run '{}' are not valid. "
               "List of valid components: {}."
               .format(self.run.run_id, VALID_COMPS))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_comps_rec_valid(self):
        """Test basic functionality of validate_comps_recorded."""
        comps = 'E1, E2, Hx, Hy, Hz'
        self.assertTrue(self.run.validate_comps_rec(comps))

    def test_validate_start_lower_than_acq_start(self):
        """Test basic functionality of validate_start."""
        data_input = self.data_stats['start']
        md_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
        self.assertFalse(self.run.validate_start(md_input, data_input))

    def test_validate_start_valid(self):
        """Test basic functionality of validate_start."""
        data_input = self.data_stats['start']
        md_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
        self.assertTrue(self.run.validate_start(md_input, data_input))

    def test_validate_end_greater_than_acq_end(self):
        """Test basic functionality of validate_end."""
        data_input = self.data_stats['end']
        md_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
        self.assertFalse(self.run.validate_end(md_input, data_input))

    def test_validate_end_valid(self):
        """Test basic functionality of validate_end."""
        data_input = self.data_stats['end']
        md_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
        self.assertTrue(self.run.validate_end(md_input, data_input))


class TestElec(unittest.TestCase):
    """Test suite for Elec data class."""

    def setUp(self):
        """Set up test fixtures"""
        self.elec = Elec(cha_num='E1', run_id='a')
        self.md_invalid = {'pos_elec_dir',
                           'neg_elec_dir'}
        self.param = ("contact resistance (start)", 0, 3000, 'Ω')

    def test_set_elec_comp_cha_undefined_dir(self):
        """Test basic functionality of set_elec_comp_cha."""
        self.elec.pos_elec_dir = None
        self.elec.neg_elec_dir = 'West'
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.elec.set_elec_comp_cha()
        msg = ("The direction of the positive and negative electrodes should "
               "be either: North, South, East or West. Please, provide "
               "electrode direction(s) for run '{0}' and channel number "
               "'{1}'! {2}".format(self.elec.run_id, self.elec.cha_num, MSG_E_CHA))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
        self.assertIsNone(self.elec.comp)
        self.assertIsNone(self.elec.cha_name)
        self.assertSetEqual(self.elec.md_invalid, self.md_invalid)

    def test_set_elec_comp_cha_invalid_dir_pair(self):
        """Test basic functionality of set_elec_comp_cha."""
        self.elec.pos_elec_dir = 'North'
        self.elec.neg_elec_dir = 'West'
        with self.assertLogs(logger, level='WARNING') as cmd:
            self.elec.set_elec_comp_cha()
        msg = ("By convention, one electrode pair is installed in a "
               "north-south direction and the other pair in a  east-west "
               "direction (check run '{0}' and channel number '{1}')!"
               .format(self.elec.run_id, self.elec.cha_num))
        self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
        self.assertIsNone(self.elec.comp)
        self.assertIsNone(self.elec.cha_name)
        self.assertSetEqual(self.elec.md_invalid, self.md_invalid)

    def test_set_elec_comp_cha_same_dir(self):
        """Test basic functionality of set_elec_comp_cha."""
        self.elec.pos_elec_dir = 'North'
        self.elec.neg_elec_dir = 'North'
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.elec.set_elec_comp_cha()
        msg = ("The direction of the positive and negative electrodes in a "
               "given pair cannot be the same (check run '{0}' and channel "
               "number '{1}')!".format(self.elec.run_id, self.elec.cha_num))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
        self.assertIsNone(self.elec.comp)
        self.assertIsNone(self.elec.cha_name)
        self.assertSetEqual(self.elec.md_invalid, self.md_invalid)

    def test_set_elec_comp_cha_valid(self):
        """Test basic functionality of set_elec_comp_cha."""
        self.elec.pos_elec_dir = 'North'
        self.elec.neg_elec_dir = 'South'
        self.elec.md_invalid = self.md_invalid
        self.elec.set_elec_comp_cha()
        self.assertEqual(self.elec.comp, 'Ex')
        self.assertEqual(self.elec.cha_name, 'LQN')
        self.assertSetEqual(self.elec.md_invalid, set())

    def test_set_elec_info_default(self):
        """Test basic functionality of set_elec_info."""
        self.elec.inst_specs = 'Borin STELTH 4 - Silver-Silver Chloride'
        self.elec.set_elec_info()
        self.assertEqual(self.elec.inst_manufacturer, "Borin")
        self.assertEqual(self.elec.inst_model, "STELTH 4")
        self.assertEqual(self.elec.inst_type, "Silver-Silver Chloride")

    def test_set_elec_info_user_defined(self):
        """Test basic functionality of set_elec_info."""
        self.elec.inst_specs = 'Manufacturer: a - Model: b - Type: c'
        self.elec.set_elec_info()
        self.assertEqual(self.elec.inst_manufacturer, "a")
        self.assertEqual(self.elec.inst_model, "b")
        self.assertEqual(self.elec.inst_type, "c")

    def test_validate_e_prop_undefined(self):
        """Test basic functionality of validate_e_prop."""
        e_prop, min_range, max_range, units = self.param
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.elec.validate_e_prop(self.param, None)
        msg = ("The {0} for run '{1}' and channel number '{2}' should be a "
               "float. {3}".format(e_prop, self.elec.run_id,
                                   self.elec.cha_num, MSG_E_CHA))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_e_prop_erroneous_type(self):
        """Test basic functionality of validate_e_prop."""
        e_prop, min_range, max_range, units = self.param
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.elec.validate_e_prop(self.param, 'a')
        msg = ("The {0} for run '{1}' and channel number '{2}' should be a "
               "float. {3}".format(e_prop, self.elec.run_id,
                                   self.elec.cha_num, MSG_E_CHA))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_e_prop_inferior_to_min_value(self):
        """Test basic functionality of validate_e_prop."""
        e_prop, min_range, max_range, units = self.param
        with self.assertLogs(logger, level='ERROR') as cmd:
            self.elec.validate_e_prop(self.param, -10)
        msg = ("The {0} for run '{1}' and for channel number '{2}' should be "
               "positive.".format(e_prop, self.elec.run_id, self.elec.cha_num))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])

    def test_validate_e_prop_superior_to_max_value(self):
        """Test basic functionality of validate_e_prop."""
        e_prop, min_range, max_range, units = self.param
        with self.assertLogs(logger, level='WARNING') as cmd:
            self.elec.validate_e_prop(self.param, 5000)
        msg = ("The {0} for run '{1}' and for channel number '{2}' should be "
               "less than {3}{4}. A {0} > {3}{4} is indicative of poor "
               "contact.".format(e_prop, self.elec.run_id,
                                 self.elec.cha_num, max_range, units))
        self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])

    def test_validate_e_prop_valid(self):
        """Test basic functionality of validate_e_prop."""
        self.assertTrue(self.elec.validate_e_prop(self.param, 500))

    def test_validate_dipole_len_empty_data_inputs(self):
        """Test basic functionality of validate_dipole_len."""
        user_input = 99
        data_input = {}
        self.assertTrue(self.elec.validate_dipole_len(user_input, data_input))

    def test_validate_dipole_len_mismatch_between_user_and_data_inputs(self):
        """Test basic functionality of validate_dipole_len."""
        user_input = 99
        data_input = {'E1': 101, 'E2': 102, 'E3': 99, 'E4': 95}
        with self.assertLogs(logger, level='ERROR') as cmd:
            valid = self.elec.validate_dipole_len(user_input, data_input)
        msg = ("The dipole length for run '{0}' and channel number '{1}' does "
               "not match the dipole length value that was configured using the "
               "LEMI424 logger in the field (99 != 101)! Update value using the "
               "GUI.".format(self.elec.run_id, self.elec.cha_num))
        self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
        self.assertFalse(valid)

    def test_validate_dipole_len_match_between_user_and_data_inputs(self):
        """Test basic functionality of validate_dipole_len."""
        user_input = 99
        data_input = {'E1': 99, 'E2': 102, 'E3': 99, 'E4': 95}
        self.assertTrue(self.elec.validate_dipole_len(user_input, data_input))


class TestMag(unittest.TestCase):
    """Test suite for Mag data class."""

    def setUp(self):
        """Set up test fixtures"""
        self.mag = Mag()

    def test_set_mag_info_default(self):
        """Test basic functionality of set_mag_info."""
        self.mag.inst_specs = 'LEMI-039'
        self.mag.set_mag_info()
        self.assertEqual(self.mag.inst_manufacturer, "LEMI LLC.")
        self.assertEqual(self.mag.inst_model, "LEMI-039")
        self.assertEqual(self.mag.inst_type,
                         "3-component analog magnetometer")

    def test_set_mag_info_user_defined(self):
        """Test basic functionality of set_mag_info."""
        self.mag.inst_specs = 'Manufacturer: a - Model: b - Type: c'
        self.mag.set_mag_info()
        self.assertEqual(self.mag.inst_manufacturer, "a")
        self.assertEqual(self.mag.inst_model, "b")
        self.assertEqual(self.mag.inst_type, "c")


class TestAux(unittest.TestCase):
    """Test suite for Aux data class."""

    def setUp(self):
        """Set up test fixtures"""
        self.aux = Aux()

    def test_set_aux_info(self):
        """Test basic functionality of set_aux_info."""
        self.aux.set_aux_info()
        self.assertEqual(self.aux.inst_manufacturer, "LEMI LLC.")
        self.assertEqual(self.aux.inst_model, "LEMI-424")
        self.assertEqual(self.aux.inst_type, "long-period 32-bit")

    def test_set_data_type(self):
        """Test basic functionality of set_data_type."""
        self.aux.set_data_type()
        self.assertEqual(self.aux.data_type, "HEALTH")

    def test_set_meas_azimuth(self):
        """Test basic functionality of set_meas_azimuth."""
        self.aux.set_meas_azimuth()
        self.assertEqual(self.aux.meas_azimuth, 0.0)

    def test_set_meas_tilt(self):
        """Test basic functionality of set_meas_tilt."""
        self.aux.set_meas_tilt()
        self.assertEqual(self.aux.meas_tilt, 0.0)