test_meteofrance.py 7.02 KB
Newer Older
import os
import shutil
import tempfile

from unittest import TestCase
from mock import Mock
import mock

from earthdiagnostics.modelingrealm import ModelingRealms
from earthdiagnostics.frequency import Frequencies
from earthdiagnostics.data_convention import MeteoFranceConvention


class TestMeteoFranceConvention(TestCase):

    def setUp(self):
        self.tmp_dir = tempfile.mkdtemp()
        os.mkdir(os.path.join(self.tmp_dir, 'expid'))
        self.config = Mock()
        self.config.data_dir = self.tmp_dir

        self.config.experiment.experiment_name = 'experiment_name'
        self.config.experiment.expid = 'expid'
        self.config.experiment.institute = 'institute'
        self.config.experiment.model = 'model'
        self.config.experiment.member_count_start = 0
        self.config.experiment.chunk_size = 1
        self.config.experiment.calendar = 'standard'

        self.config.cmor.activity = 'activity'
        self.config.cmor.append_startdate = False
        self.config.cmor.initialization_number = 1
        self.config.cmor.version = 'version'

        self.convention = MeteoFranceConvention('name', self.config)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir)

    def test_get_startdate_path(self):
        self.assertEqual(self.convention.get_startdate_path('19900101'),
                         os.path.join(self.tmp_dir, 'expid/cmorfiles/activity/institute/model/experiment_name'))

    def test_experiment_name(self):
        self.assertEqual(self.convention.experiment_name('19900101'),
                         'experiment_name')

    def test_experiment_name_append(self):
        self.config.cmor.append_startdate = True
        self.assertEqual(self.convention.experiment_name('19900101'),
                         'experiment_nameS19900101')

    def test_get_cmor_folder_path(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_cmor_folder_path('19900101', 1, ModelingRealms.ocean, 'var', 'mon',
                                                         None, cmor_var)
        self.assertEqual(file_path,
                         os.path.join(self.tmp_dir, 'experiment_name/HA/1990'))

    def test_get_cmor_folder_path_atmos(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_cmor_folder_path('19900101', 1, ModelingRealms.atmos, 'var', 'mon',
                                                         None, cmor_var)
        self.assertEqual(file_path,
                         os.path.join(self.tmp_dir, 'experiment_name/HA/1990'))

    def test_get_cmor_folder_path_custom_grid(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_cmor_folder_path('19900101', 1, ModelingRealms.ocean, 'var', 'mon',
                                                         'grid', cmor_var)
        self.assertEqual(file_path,
                         os.path.join(self.tmp_dir, 'experiment_name/HA/1990'))

    def test_get_cmor_folder_path_no_cmor(self):
        frequency = Mock()
        frequency.__str__ = Mock()
        frequency.__str__.return_value = 'mon'
        frequency.frequency = 'mon'
        file_path = self.convention.get_cmor_folder_path('19900101', 1, ModelingRealms.ocean, 'var',
                                                         Frequencies.monthly, None, None)
        self.assertEqual(file_path,
                         os.path.join(self.tmp_dir, 'experiment_name/HA/1990'))

    def test_get_file_path_no_version(self):
        self.config.cmor.version = ''
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_cmor_folder_path('19900101', 1, ModelingRealms.ocean, 'var',
                                                         Frequencies.monthly, None, None)
        self.assertEqual(file_path,
                         os.path.join(self.tmp_dir, 'experiment_name/HA/1990'))

    def test_get_filename(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', cmor_var,
                                                  Frequencies.monthly, 1, None, None, None)
        self.assertEqual(file_path,
                         'var_mon_199001_01.nc')

    def test_get_filename_daily(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', cmor_var,
                                                  Frequencies.daily, 1,
                                                  None, None, None)
        self.assertEqual(file_path,
                         'var_day_199001_01.nc')

    def test_get_filename_atmos(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Amon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_file_name('19900101', 1, ModelingRealms.atmos, 'var', cmor_var,
                                                  Frequencies.monthly, 1, None, None, None)
        self.assertEqual(file_path,
                         'var_mon_199001_01.nc')

    def test_get_filename_grid(self):
        cmor_var = Mock()
        omon = Mock()
        omon.name = 'Omon'
        cmor_var.get_table.return_value = omon
        file_path = self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', cmor_var,
                                                  Frequencies.monthly, 1, None, None, 'grid')
        self.assertEqual(file_path,
                         'var_mon_199001_01.nc')

    def test_get_filename_year(self):
        with self.assertRaises(ValueError):
            self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', None,
                                          Frequencies.yearly, None, 1990, None, None)

    def test_get_filename_date_Str(self):
        with self.assertRaises(ValueError):
            self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', None,
                                          Frequencies.yearly, None, None, 'date_str', None)

    def test_get_filename_no_date_info(self):
        with self.assertRaises(ValueError):
            self.convention.get_file_name('19900101', 1, ModelingRealms.ocean, 'var', None,
                                          Frequencies.monthly, None, None, None, None)

    def test_is_cmorized(self):
        self.assertTrue(self.convention.is_cmorized('20000101', 1, 1, ModelingRealms.ocean))

    def test_create_links_meteofrance(self):
        self.convention.create_links(None, None)

    def test_create_link_meteofrance(self):
        self.convention.create_link(None, None, None, None, None, None, None)