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 SPECSConvention
class TestSpecsConvention(TestCase):
def setUp(self):
self.tmp_dir = tempfile.mkdtemp()
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.cmor.append_startdate = False
self.config.cmor.initialization_number = 1
self.convention = SPECSConvention("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/institute/model/"
"experiment_name/S19900101",
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()
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,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S19900101/mon/ocean/var/r2i1p1/version",
def test_get_cmor_folder_path_atmos(self):
cmor_var = Mock()
omon = Mock()
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,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S19900101/mon/atmos/var/r2i1p1/version",
def test_get_cmor_folder_path_custom_grid(self):
cmor_var = Mock()
omon = Mock()
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,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S19900101/mon/ocean/var/grid/r2i1p1/version",
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,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S19900101/mon/ocean/var/r2i1p1/version",
def test_get_file_path_no_version(self):
cmor_var = Mock()
omon = Mock()
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,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S19900101/mon/ocean/var/r2i1p1",
def test_get_filename(self):
cmor_var = Mock()
omon = Mock()
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_Omon_model_experiment_name_S19900101_r2i1p1_199001-199001.nc",
)
file_path = self.convention.get_file_name(
"19900101",
1,
ModelingRealms.ocean,
"var",
None,
Frequencies.monthly,
1,
None,
None,
None,
)
self.assertEqual(
file_path,
"var_Omon_model_experiment_name_S19900101_r2i1p1_199001-199001.nc",
)
def test_get_filename_daily(self):
cmor_var = Mock()
omon = Mock()
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_Omon_model_experiment_name_S19900101_r2i1p1_199001-199001.nc",
)
def test_get_filename_atmos(self):
cmor_var = Mock()
omon = Mock()
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_Amon_model_experiment_name_S19900101_r2i1p1_199001-199001.nc",
)
def test_get_filename_grid(self):
cmor_var = Mock()
omon = Mock()
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_Omon_model_experiment_name_S19900101_r2i1p1_199001-199001.nc",
)
def test_get_filename_year(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
file_path = self.convention.get_file_name(
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
Frequencies.yearly,
None,
1990,
None,
None,
)
self.assertEqual(
file_path,
"var_Omon_model_experiment_name_S19900101_r2i1p1_1990.nc",
)
def test_get_filename_date_str(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
file_path = self.convention.get_file_name(
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
Frequencies.monthly,
None,
None,
"date_str",
None,
)
self.assertEqual(
file_path,
"var_Omon_model_experiment_name_S19900101_r2i1p1_date_str.nc",
)
def test_get_filename_no_date_info(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
with self.assertRaises(ValueError):
self.convention.get_file_name(
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
Frequencies.monthly,
None,
None,
None,
None,
)
def test_get_file_path_raise_incompatible_date_info(self):
cmor_var = Mock()
omon = Mock()
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
frequency = "mon"
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
None,
frequency,
year="1998",
)
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
1,
frequency,
year="1998",
)
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
1,
frequency,
date_str="1998",
)
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
None,
frequency,
year="1998",
date_str="1998",
)
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
1,
frequency,
year="1998",
date_str="1998",
)
self.assertRaises(
ValueError,
self.convention.get_file_path,
"19900101",
1,
ModelingRealms.ocean,
"var",
cmor_var,
None,
frequency,
)
@mock.patch("os.path.isfile")
def test_is_cmorized(self, mock_is_file):
mock_is_file.return_value = True
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 1
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101/mon/ocean/var/r2i1p1/version",
)
)
self.assertTrue(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
@mock.patch("os.path.isfile")
def test_is_not_cmorized(self, mock_is_file):
mock_is_file.return_value = False
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 1
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101/mon/ocean/var/r2i1p1/version",
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
@mock.patch("os.path.isfile")
def test_is_cmorized_no_startdate_path(self, mock_is_file):
mock_is_file.return_value = False
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 1
self.tmp_dir,
"expid/original_files/cmorfiles/institute/model/"
"experiment_name"
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
def test_is_cmorized_false_not_member_folder(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 1
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101/mon/ocean/var/",
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
@mock.patch("os.path.isfile")
def test_is_cmorized_not_enough_vars(self, mock_is_file):
mock_is_file.return_value = True
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 2
os.makedirs(
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101/mon/ocean/var/r2i1p1/version",
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
def test_is_cmorized_not_domain_folder(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 2
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101/mon/r2i1p1/version",
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
def test_is_cmorized_not_freq_folder(self):
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
self.config.var_manager.get_variable.return_value = cmor_var
self.config.cmor.min_cmorized_vars = 2
"expid/original_files/cmorfiles/institute/model/"
"experiment_name/S20000101",
)
)
self.assertFalse(
self.convention.is_cmorized("20000101", 1, 1, ModelingRealms.ocean)
)
@mock.patch("earthdiagnostics.data_convention.SPECSConvention.create_link")
def test_create_links(self, mock_create_link):
"expid/original_files/cmorfiles/institute/model/experiment_name/"
"S20010101/mon/ocean/var/r2i1p1",
os.makedirs(member_path)
tempfile.mkstemp(dir=member_path)
@mock.patch("earthdiagnostics.data_convention.SPECSConvention.create_link")
def test_create_links_member_not_found(self, mock_create_link):
"expid/original_files/cmorfiles/institute/model/experiment_name/"
"S20010101/mon/ocean/var/r1i1p1",
os.makedirs(member_path)
tempfile.mkstemp(dir=member_path)
@mock.patch("earthdiagnostics.data_convention.SPECSConvention.create_link")
def test_create_links_with_grid(self, mock_create_link):
"expid/original_files/cmorfiles/institute/model/experiment_name/"
"S20010101/mon/ocean/var/r2i1p1/grid",
os.makedirs(member_path)
tempfile.mkstemp(dir=member_path)