Skip to content

Modernize tests for esmvalcore.preprocessor.save #1799

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
schlunma merged 2 commits into from
Nov 22, 2022
Merged

Modernize tests for esmvalcore.preprocessor.save #1799

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6beb689
Modernize and add tests for esmvalcore.preprocessor.save
bouweandela Nov 15, 2022
8ad58df
Remove class
bouweandela Nov 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
250 changes: 124 additions & 126 deletions tests/integration/preprocessor/_io/test_save.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,137 +1,135 @@
"""Integration tests for :func:`esmvalcore.preprocessor.save`"""

import os
import tempfile
import unittest

import iris
import netCDF4
import numpy as np
import pytest
from iris.coords import DimCoord
from iris.cube import Cube, CubeList

from esmvalcore.preprocessor import save


class TestSave(unittest.TestCase):
"""Tests for :func:`esmvalcore.preprocessor.save`"""

def setUp(self):
"""Prepare tests."""
self.temp_files = []

def tearDown(self):
for temp_file in self.temp_files:
if os.path.isfile(temp_file):
os.remove(temp_file)

def _create_sample_cube(self):
lat = DimCoord(
np.asarray([1, 2], np.single),
standard_name='latitude',
units='degrees_north')
lon = DimCoord(
np.asarray([1, 2], np.single),
standard_name='longitude',
units='degrees_east')
time = DimCoord(
np.asarray([1, 2], np.single),
standard_name='time',
units='days since 2000-1-1')

cube = Cube(
np.random.random_sample([2, 2, 2]),
var_name='sample',
units='1',
dim_coords_and_dims=((lat, 0), (lon, 1), (time, 2)))

descriptor, filename = tempfile.mkstemp('.nc')
os.close(descriptor)
self.temp_files.append(filename)
return cube, filename

def test_save(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename)
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)

def test_save_alias(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, alias='alias')
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
self.assertEqual(loaded_cube.var_name, 'alias')

def test_save_zlib(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, compress=True)
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
handler = netCDF4.Dataset(path, 'r')
sample_filters = handler.variables['sample'].filters()
self.assertTrue(sample_filters['zlib'])
self.assertTrue(sample_filters['shuffle'])
self.assertEqual(sample_filters['complevel'], 4)
handler.close()

def test_fail_empty_cubes(self):
"""Test save fails if empty cubes is provided."""
(_, filename) = self._create_sample_cube()
empty_cubes = CubeList([])
with self.assertRaises(ValueError):
save(empty_cubes, filename)

def test_fail_without_filename(self):
"""Test save fails if filename is not provided."""
cube, _ = self._create_sample_cube()
with self.assertRaises(TypeError):
save([cube])

def test_save_optimized_map(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, optimize_access='map')
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
self._check_chunks(path, [2, 2, 1])

def test_save_optimized_timeseries(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, optimize_access='timeseries')
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
self._check_chunks(path, [1, 1, 2])

def test_save_optimized_lat(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, optimize_access='latitude')
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
expected_chunks = [2, 1, 1]
self._check_chunks(path, expected_chunks)

def _check_chunks(self, path, expected_chunks):
handler = netCDF4.Dataset(path, 'r')
@pytest.fixture
def filename(tmp_path):
return tmp_path / 'test.nc'


@pytest.fixture
def cube():
lat = DimCoord(
np.asarray([1, 2], np.single),
standard_name='latitude',
units='degrees_north',
)
lon = DimCoord(
np.asarray([1, 2], np.single),
standard_name='longitude',
units='degrees_east',
)
time = DimCoord(
np.asarray([1, 2], np.single),
standard_name='time',
units='days since 2000-1-1',
)

cube = Cube(
np.random.random_sample([2, 2, 2]),
var_name='sample',
units='1',
dim_coords_and_dims=((lat, 0), (lon, 1), (time, 2)),
)

return cube


def _compare_cubes(cube, loaded_cube):
np.testing.assert_equal(cube.data, loaded_cube.data)
for coord in cube.coords():
np.testing.assert_equal(coord.points,
loaded_cube.coord(coord.name()).points)


def _check_chunks(path, expected_chunks):
with netCDF4.Dataset(path, 'r') as handler:
chunking = handler.variables['sample'].chunking()
handler.close()
self.assertListEqual(expected_chunks, chunking)

def test_save_optimized_lon_time(self):
"""Test save"""
cube, filename = self._create_sample_cube()
path = save([cube], filename, optimize_access='longitude time')
loaded_cube = iris.load_cube(path)
self._compare_cubes(cube, loaded_cube)
self._check_chunks(path, [1, 2, 2])

def _compare_cubes(self, cube, loaded_cube):
self.assertTrue((cube.data == loaded_cube.data).all())
for coord in cube.coords():
self.assertTrue(
(coord.points == loaded_cube.coord(coord.name()).points).all())
assert expected_chunks == chunking


def test_save(cube, filename):
"""Test save."""
path = save([cube], filename)
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)


def test_save_create_parent_dir(cube, tmp_path):
filename = tmp_path / 'preproc' / 'something' / 'test.nc'
path = save([cube], filename)
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)


def test_save_alias(cube, filename):
"""Test save."""
path = save([cube], filename, alias='alias')
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
assert loaded_cube.var_name == 'alias'


def test_save_zlib(cube, filename):
"""Test save."""
path = save([cube], filename, compress=True)
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
with netCDF4.Dataset(path, 'r') as handler:
sample_filters = handler.variables['sample'].filters()
assert sample_filters['zlib'] is True
assert sample_filters['shuffle'] is True
assert sample_filters['complevel'] == 4


def test_fail_empty_cubes(filename):
"""Test save fails if empty cubes is provided."""
empty_cubes = CubeList([])
with pytest.raises(ValueError):
save(empty_cubes, filename)


def test_fail_without_filename(cube):
"""Test save fails if filename is not provided."""
with pytest.raises(TypeError):
save([cube])


def test_save_optimized_map(cube, filename):
"""Test save."""
path = save([cube], filename, optimize_access='map')
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
_check_chunks(path, [2, 2, 1])


def test_save_optimized_timeseries(cube, filename):
"""Test save."""
path = save([cube], filename, optimize_access='timeseries')
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
_check_chunks(path, [1, 1, 2])


def test_save_optimized_lat(cube, filename):
"""Test save."""
path = save([cube], filename, optimize_access='latitude')
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
expected_chunks = [2, 1, 1]
_check_chunks(path, expected_chunks)


def test_save_optimized_lon_time(cube, filename):
"""Test save."""
path = save([cube], filename, optimize_access='longitude time')
loaded_cube = iris.load_cube(path)
_compare_cubes(cube, loaded_cube)
_check_chunks(path, [1, 2, 2])