Skip to content
Snippets Groups Projects
test_lemi_metadata.py 14.9 KiB
Newer Older
#!/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, log_file_path)

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