#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Tests for `lemi_metadata` module."""
import logging
import unittest
from obspy import UTCDateTime
from pathlib import Path
from lemi2seed.lemi_data import LemiData, VALID_COMPONENTS
from lemi2seed.lemi_metadata import (BaseSurvey, Survey, BaseStation, Station_,
Run, Electric, log_file_path)
from lemi2seed.utils import MSG_E_CHA
OUTPUT_MSEED = Path(__file__).parent.joinpath('MSEED')
TEST_DIR = Path(__file__).parent.joinpath('test_data')
SCR_DIR = "lemi2seed.lemi_metadata"
logging.config.fileConfig(log_file_path)
logger = logging.getLogger(SCR_DIR)
class TestBaseSurvey(unittest.TestCase):
"""Test suite for BaseSurvey data class."""
def setUp(self):
"""Set up test fixtures."""
lemi_data = LemiData(TEST_DIR.joinpath('DATA0110'), OUTPUT_MSEED)
lemi_data.prep_data()
self.data_stats = lemi_data.stats
def test_validate_time_period_start_not_utc(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = 2021.167
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_greater_than_acquisition_start(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_valid(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
bs = BaseSurvey()
self.assertTrue(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_end_not_utc(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = 2021.167
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_lower_than_acquisition_end(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_valid(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
bs = BaseSurvey()
self.assertTrue(bs.validate_time_period_end(metadata_input, data_input))
class TestSurvey(unittest.TestCase):
"""Test suite for Survey data class."""
def test_validate_archive_network_undefined(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network(None)
msg = "The network code should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_erroneous_type(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network(12)
msg = "The network code should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_invalid(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network('EMX')
msg = "The network code should be two alphanumeric character long."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_valid(self):
"""Test basic functionality of validate_archive_network."""
self.assertTrue(Survey.validate_archive_network('EM'))
def test_validate_citation_dataset_doi_undefined(self):
"""Test basic functionality of validate_citation_dataset_doi."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_citation_dataset_doi(None)
msg = "The DOI number(s) should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
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:
Survey.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:
Survey.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:
Survey.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(Survey.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(Survey.validate_citation_dataset_doi(dois))
def test_validate_project_lead_email_undefined(self):
"""Test basic functionality of validate_project_lead_email."""
self.assertTrue(Survey.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:
Survey.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(Survey.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(Survey.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(Survey.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(Survey.validate_project_lead_email(emails))
class TestBaseStation(unittest.TestCase):
"""Test suite for BaseStation 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, 'm', '')
self.param = param
def test_validate_geographics_undefined(self):
"""Test basic functionality of validate_geographics."""
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, None)
msg = "Station elevation should be a float. "
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_geographics_erroneous_type(self):
"""Test basic functionality of validate_geographics."""
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, 'a')
msg = "Station elevation should be a float. "
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_geographics_not_in_range(self):
"""Test basic functionality of validate_geographics."""
geo, min_range, max_range, min_val, max_val, units, msg = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(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_geographics_do_not_match_logger_recorded_metadata(self):
"""Test basic functionality of validate_geographics."""
geo = 'Station elevation'
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, 2250)
msg = ("Unexpected {0}! Provided {0} should roughly match the {0} "
"recorded by the on-site GPS.".format(geo))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
class TestStation_(unittest.TestCase):
"""Test suite for Station_ data class."""
def test_validate_archive_id_undefined(self):
"""Test basic functionality of validate_archive_id."""
with self.assertLogs(logger, level='ERROR') as cmd:
Station_.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:
Station_.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:
Station_.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(Station_.validate_archive_id('KELLY'))
def test_validate_submitter_email_undefined(self):
"""Test basic functionality of validate_submitter_email."""
self.assertTrue(Station_.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('DATA0110'), OUTPUT_MSEED)
lemi_data.prep_data()
self.data_stats = lemi_data.stats
def test_validate_components_recorded_undefined(self):
"""Test basic functionality of validate_components_recorded."""
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(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_components_recorded_erroneous_type(self):
"""Test basic functionality of validate_components_recorded."""
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(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_components_recorded_invalid(self):
"""Test basic functionality of validate_components_recorded."""
components = 'E1, E2, Hx, Hy, Hn'
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(components)
msg = ("Some of the listed components for run '{}' are not valid. "
"List of valid components: {}."
.format(self.run.run_id, VALID_COMPONENTS))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_components_recorded_valid(self):
"""Test basic functionality of validate_components_recorded."""
components = 'E1, E2, Hx, Hy, Hz'
self.assertTrue(self.run.validate_components_recorded(components))
def test_validate_time_period_start_lower_than_acquisition_start(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
self.assertFalse(self.run.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_valid(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
self.assertTrue(self.run.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_end_greater_than_acquisition_end(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
self.assertFalse(self.run.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_valid(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
self.assertTrue(self.run.validate_time_period_end(metadata_input, data_input))
class TestElectric(unittest.TestCase):
"""Test suite for Electric data class."""
def setUp(self):
"""Set up test fixtures"""
self.elec = Electric(channel_number='E1', run_id='a')
self.metadata_invalid = {'positive_electrode_direction',
'negative_electrode_direction'}
self.param = ("contact resistance (start)", 0, 3000, 'Ω')
def test_set_electric_component_channel_undefined_direction(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = None
self.elec.negative_electrode_direction = 'West'
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.set_electric_component_channel()
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.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_invalid_direction_pair(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'West'
with self.assertLogs(logger, level='WARNING') as cmd:
self.elec.set_electric_component_channel()
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.channel_number))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_same_direction(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'North'
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.set_electric_component_channel()
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.channel_number))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_valid(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'South'
self.elec.metadata_invalid = self.metadata_invalid
self.elec.set_electric_component_channel()
self.assertEqual(self.elec.component, 'Ex')
self.assertEqual(self.elec.channel_name, 'LQN')
self.assertSetEqual(self.elec.metadata_invalid, set())
def test_set_electrode_info_default(self):
"""Test basic functionality of set_electrode_info."""
self.elec.instrument_specs = 'Borin STELTH 4 - Silver-Silver Chloride'
self.elec.set_electrode_info()
self.assertEqual(self.elec.instrument_manufacturer, "Borin")
self.assertEqual(self.elec.instrument_model, "STELTH 4")
self.assertEqual(self.elec.instrument_type, "Silver-Silver Chloride")
def test_set_electrode_info_user_defined(self):
"""Test basic functionality of set_electrode_info."""
self.elec.instrument_specs = 'Manufacturer: a - Model: b - Type: c'
self.elec.set_electrode_info()
self.assertEqual(self.elec.instrument_manufacturer, "a")
self.assertEqual(self.elec.instrument_model, "b")
self.assertEqual(self.elec.instrument_type, "c")
def test_validate_e_property_undefined(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(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.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_erroneous_type(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(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.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_inferior_to_min_value(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(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.channel_number))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_superior_to_max_value(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='WARNING') as cmd:
self.elec.validate_e_property(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.channel_number, max_range, units))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
def test_validate_e_property_valid(self):
"""Test basic functionality of validate_e_property."""
self.assertTrue(self.elec.validate_e_property(self.param, 500))