test_data_convention.py

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()
        os.mkdir(os.path.join(self.tmp_dir, "expid"))

        self.config = Mock()
        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.config.cmor.append_startdate = False
        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
            )

    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/expname",
            ),
        )

    def test_experiment_name(self):
        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"
        cmor_var = Mock()
        omon = Mock()
        omon.name = "Omon"
        cmor_var.get_table.return_value = omon

        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()
        omon.name = "Omon"
        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
        )
        filename = os.path.basename(path)
        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"
        with self.assertRaises(ValueError):
            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"
        filename = os.path.basename(path)

        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 = Mock()
        frequency.folder_name.return_value = "frequency_folder"
        self.convention.create_link(
            ModelingRealms.ocean, path, frequency, "var", None, False, None
        )
        filename = os.path.basename(path)
        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",
        )
        fd.close()

        frequency = Mock()
        frequency.folder_name.return_value = "frequency_folder"
        self.convention.create_link(
            ModelingRealms.ocean, path, frequency, "var", None, True, None
        )
        filename = os.path.basename(path)
        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
        )
        filename = os.path.basename(path)
        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 = Mock()
        frequency.folder_name.return_value = "frequency_folder"
        self.convention.create_link(
            ModelingRealms.ocean, path, frequency, "var", "grid", False, None
        )
        filename = os.path.basename(path)
        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 = Mock()
        frequency.folder_name.return_value = "frequency_folder"
        self.convention.create_link(
            ModelingRealms.ocean, path, frequency, "var", "grid", False, None
        )
        filename = os.path.basename(path)
        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)