import os import shutil import tempfile from unittest import TestCase from 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/original_files/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)