Newer
Older
import os
import shutil
from unittest import TestCase
from mock import Mock
import mock
import tempfile
from earthdiagnostics.data_convention import DataConvention
from earthdiagnostics.modelingrealm import ModelingRealms
class TestDataConvention(TestCase):
def setUp(self):
self.tmp_dir = tempfile.mkdtemp()
self.config.frequency = "mon"
self.config.experiment.experiment_name = "expname"
self.config.data_dir = self.tmp_dir
self.config.experiment.expid = "expid"
self.config.cmor.activity = "activity"
self.config.experiment.institute = "institute"
self.config.experiment.model = "model"
self.config.experiment.atmos_timestep = 3
self.config.experiment.ocean_timestep = 6
self.convention = DataConvention("name", self.config)
def tearDown(self):
shutil.rmtree(self.tmp_dir)
def test_get_filename(self):
with self.assertRaises(NotImplementedError):
self.convention.get_file_name(
None, None, None, None, None, None, None, None, None, None
)
self.assertEqual(
self.convention.get_startdate_path("19900101"),
os.path.join(
self.tmp_dir,
"expid/original_files/cmorfiles/activity/institute/model/"
"expname",
self.assertEqual(
self.convention.experiment_name("19900101"), "expname"
)
def test_experiment_name_append(self):
self.config.cmor.append_startdate = True
self.assertEqual(
self.convention.experiment_name("19900101"), "expnameS19900101"
)
def test_get_cmor_folder_path(self):
with self.assertRaises(NotImplementedError):
self.convention.get_cmor_folder_path(
None, None, None, None, None, None, None
)
def test_create_links(self):
with self.assertRaises(NotImplementedError):
self.convention.create_links(None, None)
def test_get_member_str(self):
with self.assertRaises(NotImplementedError):
self.convention.get_member_str(None)
def test_is_cmorized(self):
with self.assertRaises(NotImplementedError):
self.convention.is_cmorized(None, None, None, None)
@mock.patch(
"earthdiagnostics.data_convention.DataConvention.get_cmor_folder_path",
autospec=True,
)
@mock.patch(
"earthdiagnostics.data_convention.DataConvention.get_file_name",
autospec=True,
)
def test_get_file_path(self, mock_filename, mock_cmor_path):
mock_cmor_path.return_value = "path"
mock_filename.return_value = "filename"
filepath = self.convention.get_file_path(
"19900101", 1, ModelingRealms.ocean, "var", cmor_var, 1, "mon"
)
self.assertEqual("path/filename", filepath)
@mock.patch(
"earthdiagnostics.data_convention.DataConvention.get_cmor_folder_path",
autospec=True,
)
@mock.patch(
"earthdiagnostics.data_convention.DataConvention.get_file_name",
autospec=True,
)
def test_get_file_path_no_freq(self, mock_filename, mock_cmor_path):
mock_cmor_path.return_value = "path"
mock_filename.return_value = "filename"
cmor_var = Mock()
omon = Mock()
cmor_var.get_table.return_value = omon
filepath = self.convention.get_file_path(
"19900101", 1, ModelingRealms.ocean, "var", cmor_var, 1, None
)
self.assertEqual("path/filename", filepath)
def test_create_link(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
frequency = Mock()
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", None, False, None
)
self.assertTrue(
os.path.islink(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"var_f6h",
filename,
)
)
)
def test_create_link_no_source(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
os.remove(path)
frequency = Mock()
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", None, False, None
)
def test_create_link_exists(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
frequency = Mock()
frequency.folder_name.return_value = "frequency_folder"
os.makedirs(
os.path.join(self.tmp_dir, "expid", "frequency_folder", "var_f6h")
)
os.symlink(
path,
os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var_f6h", filename
),
)
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", None, False, None
)
self.assertTrue(
os.path.islink(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"var_f6h",
filename,
)
)
)
def test_create_link_default_is_link(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
os.makedirs(
os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-original_f6h"
)
)
os.symlink(
os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-original_f6h"
),
os.path.join(self.tmp_dir, "expid", "frequency_folder", "var_f6h"),
)
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", None, False, None
)
self.assertTrue(
os.path.islink(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"var_f6h",
filename,
)
)
)
def test_create_link_move_old(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
os.makedirs(
os.path.join(self.tmp_dir, "expid", "frequency_folder", "var_f6h")
)
fd = open(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"var_f6h",
"var_20001101.nc",
),
"w",
)
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", None, True, None
)
self.assertTrue(
os.path.islink(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"var_f6h",
filename,
)
)
)
self.assertTrue(
os.path.isfile(
os.path.join(
self.tmp_dir,
"expid",
"frequency_folder",
"old_var_f6h",
"var_20001101.nc",
)
)
)
def test_create_link_with_grid(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
frequency = Mock()
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", "grid", False, None
)
var_mainfolder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var_f6h"
)
var_grid_folder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-grid_f6h"
)
self.assertTrue(os.path.islink(var_mainfolder))
self.assertTrue(
os.path.islink(os.path.join(var_grid_folder, filename))
)
self.assertTrue(os.readlink(var_mainfolder), var_grid_folder)
def test_create_link_with_grid_default_is_link(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
os.makedirs(
os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-original_f6h"
)
)
os.symlink(
os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-original_f6h"
),
os.path.join(self.tmp_dir, "expid", "frequency_folder", "var_f6h"),
)
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", "grid", False, None
)
var_mainfolder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var_f6h"
)
var_grid_folder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-grid_f6h"
)
self.assertTrue(os.path.islink(var_mainfolder))
self.assertTrue(
os.path.islink(os.path.join(var_grid_folder, filename))
)
self.assertTrue(os.readlink(var_mainfolder), var_grid_folder)
def test_create_link_with_grid_default_is_folder(self):
file_descriptor, path = tempfile.mkstemp(dir=self.tmp_dir)
os.close(file_descriptor)
os.makedirs(
os.path.join(self.tmp_dir, "expid", "frequency_folder", "var_f6h")
)
frequency.folder_name.return_value = "frequency_folder"
self.convention.create_link(
ModelingRealms.ocean, path, frequency, "var", "grid", False, None
)
var_mainfolder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var_f6h"
)
var_grid_folder = os.path.join(
self.tmp_dir, "expid", "frequency_folder", "var-grid_f6h"
)
self.assertTrue(os.path.islink(var_mainfolder))
self.assertTrue(
os.path.islink(os.path.join(var_grid_folder, filename))
)
self.assertTrue(os.readlink(var_mainfolder), var_grid_folder)