Skip to content
GitLab
Commit 62ad8af0 authored by BorjaEst's avatar BorjaEst
Browse files

First stable pytest tests version

parent 8338a806
Hide whitespace changes
Inline Side-by-side
tests/__init__.py
View file @ 62ad8af0
"""Unit test package for torch_cd_example."""
import os
from o3skim import utils
data_dir = "./tests/data"
# Generation of mockup data
os.makedirs(data_dir, exist_ok=True)
with utils.cd(data_dir):
from tests import mockup_data
tests/mockup/coordinate.py
View file @ 62ad8af0
......@@ -14,7 +14,8 @@ def __timeline(start, end=None, step=10):
start += step
def time(start=2000, end=2010, step=30):
def time(start=2000, end=None, step=30):
end = start if not end else None
return ('time', [t for t in __timeline(start, end, step)])
......
tests/mockup_data.py
View file @ 62ad8af0
......@@ -6,76 +6,72 @@ from o3skim import utils
from tests import mockup
from tests.mockup import coordinate
model1_dir = "./variable_per_file"
os.makedirs(model1_dir, exist_ok=True)
with utils.cd(model1_dir):
path = "toz"
os.makedirs(path, exist_ok=True)
with utils.cd(path):
for year in range(2000, 2002):
mockup.Dataset(
var_coords={
'toz': [
coordinate.time(year),
coordinate.lat(),
coordinate.lon()
]
},
coords=[
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
).to_netcdf("toz_" + str(year) + ".nc")
path = "vmro3"
os.makedirs(path, exist_ok=True)
with utils.cd(path):
for year in range(2000, 2002):
mockup.Dataset(
var_coords={
'vmro3': [
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
},
coords=[
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
).to_netcdf("vmro3_" + str(year) + ".nc")
model2_dir = "./variables_merged"
os.makedirs(model2_dir, exist_ok=True)
with utils.cd(model2_dir):
for year in range(2000, 2002):
mockup.Dataset(
var_coords={
'toz': [
coordinate.time(year),
coordinate.lat(),
coordinate.lon()
],
'vmro3': [
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
},
coords=[
def _tco3(year, name='tco3'):
return mockup.Dataset(
var_coords={
name: [
coordinate.time(year),
coordinate.lat(),
coordinate.lon()
]
},
coords=[
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
)
def _vmro3(year, name='vmro3'):
return mockup.Dataset(
var_coords={
name: [
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
).to_netcdf("all_" + str(year) + ".nc")
},
coords=[
coordinate.time(year),
coordinate.plev(),
coordinate.lat(),
coordinate.lon()
]
)
def tco3(year_range, name='tco3'):
for year in year_range:
_tco3(year, name).to_netcdf(name + "_" + str(year) + ".nc")
def vmro3(year_range, name='vmro3'):
for year in year_range:
_vmro3(year, name).to_netcdf(name + "_" + str(year) + ".nc")
_datasets = {
'tco3': _tco3,
'vmro3': _vmro3
}
def combined(year_range, vars=['tco3', 'vmro3']):
for year in year_range:
dataset = mockup.Dataset({}, [])
for var in vars:
load = _datasets[var](year)
dataset = dataset.merge(load)
dataset.to_netcdf("merged_" + str(year) + ".nc")
def noise(name='noise'):
"""Some noise that blocks '*' wild card"""
mockup.Dataset(
var_coords={'noise': [coordinate.time(2000, 2010)]},
coords=[coordinate.time(2000, 2010)]
).to_netcdf(name + ".nc")
tests/mockup_noise.py deleted 100644 → 0
View file @ 8338a806
"""This modules creates mockup data for testing"""
import xarray as xr
import numpy as np
import datetime
import random
base = datetime.datetime(2000, 1, 1)
indexes = {
'time': [base + datetime.timedelta(days=9*i) for i in range(99)],
'plev': [x for x in range(1, 1000, 100)],
'lat': [x for x in range(-90, 90, 10)],
'lon': [x for x in range(-180, 180, 20)]
}
def data_vars():
"""Creates a mock n-array with coordinate values"""
dim = [len(axis) for _, axis in indexes.items()]
return tuple(indexes), np.ones(dim),
def data_coord():
"""Creates a mock coordinates"""
return indexes
def dataset(name):
"""Creates a dataset according to the global module indexes"""
return xr.Dataset(
{name: data_vars()},
coords = data_coord()
)
def netcdf(path, name, **kwarg):
"""Creates or appends data to a noise netcdf file"""
ds=dataset(name)
ds.to_netcdf(path)
tests/noise_files.yaml deleted 100644 → 0
View file @ 8338a806
- name: toz
path: Ccmi/mon/toz/toz_noise.nc
- name: tco3
path: Ecmwf/Era5/tco3_noise.nc
- name: toz
path: Ecmwf/Erai/toz_noise.nc
- name: vmro3
path: Ecmwf/Erai/vmro3_noise.nc
tests/sources_err.yaml
View file @ 62ad8af0
ErrorModels:
correct_variable:
vmro3_zm: # Correct variable
name: vmro3
paths: Ecmwf/Erai/vmro3_????.nc
SourceSplit:
ModelBadName:
tco3_zm:
name: NonExistingName
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
plev: level
lat: latitude
lon: longitude
non_existing_variable:
tco3_zm: # Incorrect variable
name: non_existing_var
paths: Ecmwf/Erai/toz_????.nc
lat: lat
lon: lon
ModelBadPath:
tco3_zm:
name: tco3
paths: SourceSplit/no_file_????.nc
coordinates:
time: time
lat: latitude
lon: longitude
lat: lat
ModelMissingCoords:
tco3_zm:
name: tco3
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
lat: lat
ModelBadCoords:
tco3_zm:
name: tco3
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
lat: lat
longitude: lon
ModelExtraCoords:
tco3_zm:
name: tco3
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
plev: plev
lat: lat
lon: lon
tests/sources_base.yaml → tests/sources_example.yaml
View file @ 62ad8af0
CCMI-1:
IPSL:
SourceSplit:
ModelTCO3:
tco3_zm:
name: toz
paths: Ccmi/mon/toz/toz_????.nc
name: tco3
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
lat: lat
lon: lon
ModelVMRO3:
vmro3_zm:
name: vmro3
paths: Ccmi/mon/vmro3/*.nc
paths: SourceSplit/vmro3_????.nc
coordinates:
time: time
plev: plev
lat: lat
lon: lon
ECMWF:
ERA-5:
ModelALL:
tco3_zm:
name: tco3
paths: SourceSplit/tco3_????.nc
coordinates:
time: time
lat: lat
lon: lon
vmro3_zm:
name: vmro3
paths: SourceSplit/vmro3_????.nc
coordinates:
time: time
plev: plev
lat: lat
lon: lon
SourceMerged:
ModelTCO3:
tco3_zm:
name: tco3
paths: Ecmwf/Era5/tco3_????.nc
paths: SourceMerged/merged_????.nc
coordinates:
time: time
lat: lat
lon: lon
ModelVMRO3:
vmro3_zm:
name: vmro3
paths: SourceMerged/merged_????.nc
coordinates:
lon: longitude
lat: latitude
time: time
ERA-i:
plev: plev
lat: lat
lon: lon
ModelALL:
tco3_zm:
name: toz
paths: Ecmwf/Erai/toz_????.nc
name: tco3
paths: SourceMerged/merged_????.nc
coordinates:
time: time
lat: latitude
lon: longitude
lat: lat
lon: lon
vmro3_zm:
name: vmro3
paths: Ecmwf/Erai/vmro3_????.nc
paths: SourceMerged/merged_????.nc
coordinates:
time: time
plev: level
lat: latitude
lon: longitude
plev: plev
lat: lat
lon: lon
tests/test_errors.py 0 → 100644
View file @ 62ad8af0
"""Unittest module template."""
import os
import pytest
import o3skim
import xarray
import tests.mockup_data as mockup_data
# Test configurations ----------------------------------------------
configuration_file = "tests/sources_err.yaml"
year_line = range(2000, 2022)
sources = ["SourceSplit"]
# models = ["ModelBadName", "ModelBadPath", "ModelMissingCoords",
# "ModelBadCoords", "ModelExtraCoords"]
config = o3skim.utils.load(configuration_file)
# Module fixtures ---------------------------------------------------
@ pytest.fixture(scope='module')
def data_dir(tmpdir_factory):
data_dir = tmpdir_factory.mktemp("data")
for source in sources:
source_dir = data_dir.join(source)
os.mkdir(source_dir)
with o3skim.utils.cd(source_dir):
if source == 'SourceMerged':
mockup_data.combined(year_line)
if source == 'SourceSplit':
mockup_data.tco3(year_line)
mockup_data.vmro3(year_line)
with o3skim.utils.cd(data_dir):
yield data_dir
# Source Tests ------------------------------------------------------
@ pytest.fixture(scope='module')
def source_config(request, data_dir):
source_name = request.param
yield source_name, config[source_name]
@ pytest.fixture(scope='class')
def source(source_config):
source_name, collections = source_config
yield o3skim.Source(source_name, collections)
@ pytest.mark.parametrize('source_config', sources, indirect=True)
class Test_Exceptions:
def test_constructor(self, source_config):
source_name, collections = source_config
source = o3skim.Source(source_name, collections)
assert source.name == source_name
assert source.models == []
assert True
def test_skimming(self, source):
assert None == source.skim(groupby=None)
assert None == source.skim(groupby='year')
assert None == source.skim(groupby='decade')
assert True
tests/test_o3skim.py
View file @ 62ad8af0
"""Unittest module template."""
import os
import shutil
import unittest
import glob
import pytest
import o3skim
import xarray as xr
from tests import mockup_data
from tests import mockup_noise
class TestO3SKIM_sources(unittest.TestCase):
"""Tests for `sources` package."""
def setUp(self):
"""Loads and creates the test folders and files from test_sources.yaml"""
self.config_base = o3skim.utils.load("tests/sources_base.yaml")
self.assertTrue(type(self.config_base) is dict)
self.config_err = o3skim.utils.load("tests/sources_err.yaml")
self.assertTrue(type(self.config_err) is dict)
self.create_mock_datasets()
self.backup_datasets()
self.assert_with_backup()
self.create_noise_datasets()
def tearDown(self):
"""Tear down test fixtures, if any."""
def create_mock_datasets(self):
"""Creates mock data files according to the loaded configuration"""
with o3skim.utils.cd('data'):
for _, collection in self.config_base.items():
for _, variables in collection.items():
for _, vinfo in variables.items():
dirname = os.path.dirname(vinfo['paths'])
os.makedirs(dirname, exist_ok=True)
mockup_data.netcdf(dirname, **vinfo)
def create_noise_datasets(self):
"""Creates noise data files according to the noise configuration"""
config_noise = o3skim.utils.load("tests/noise_files.yaml")
with o3skim.utils.cd('data'):
for ninfo in config_noise:
mockup_noise.netcdf(**ninfo)
def clean_output(self):
"""Cleans output removing all folders at output"""
with o3skim.utils.cd('output'):
directories = (d for d in os.listdir() if os.path.isdir(d))
for directory in directories:
shutil.rmtree(directory)
def backup_datasets(self):
"""Loads the mock datasets into an internal variable"""
self.ds_backup = {}
with o3skim.utils.cd('data'):
for source, collection in self.config_base.items():
self.ds_backup[source] = {}
for model, variables in collection.items():
self.ds_backup[source][model] = {}
for v, vinfo in variables.items():
with xr.open_mfdataset(vinfo['paths']) as ds:
self.ds_backup[source][model][v] = ds
def assert_with_backup(self):
"""Asserts the dataset in the backup is equal to the config load"""
with o3skim.utils.cd('data'):
for source, collection in self.config_base.items():
for model, variables in collection.items():
for v, vinfo in variables.items():
with xr.open_mfdataset(vinfo['paths']) as ds:
xr.testing.assert_identical(
self.ds_backup[source][model][v], ds)
def test_001_SourcesFromConfig(self):
"""Creates the different sources from the configuration file"""
with o3skim.utils.cd("data"):
ds = {name: o3skim.Source(name, collection) for
name, collection in self.config_base.items()}
# CCMI-1 tco3_zm asserts
self.assertTrue( 'time' in ds['CCMI-1']._models['IPSL']['tco3_zm'].coords)
self.assertTrue( 'lat' in ds['CCMI-1']._models['IPSL']['tco3_zm'].coords)
self.assertFalse( 'lon' in ds['CCMI-1']._models['IPSL']['tco3_zm'].coords)
# CCMI-1 vmro3_zm asserts
self.assertTrue( 'time' in ds['CCMI-1']._models['IPSL']['vmro3_zm'].coords)
self.assertTrue( 'plev' in ds['CCMI-1']._models['IPSL']['vmro3_zm'].coords)
self.assertTrue( 'lat' in ds['CCMI-1']._models['IPSL']['vmro3_zm'].coords)
self.assertFalse( 'lon' in ds['CCMI-1']._models['IPSL']['vmro3_zm'].coords)
# Checks the original data has not been modified
self.assert_with_backup()
def test_002_OutputFromSources(self):
"""Skims the data into the output folder"""
with o3skim.utils.cd("data"):
ds = {name: o3skim.Source(name, collection) for
name, collection in self.config_base.items()}
with o3skim.utils.cd("output"):
[source.skim() for source in ds.values()]
# CCMI-1 data skim asserts
self.assertTrue(os.path.isdir("output/CCMI-1_IPSL"))
self.assertTrue(os.path.exists("output/CCMI-1_IPSL/tco3_zm.nc"))
self.assertTrue(os.path.exists("output/CCMI-1_IPSL/vmro3_zm.nc"))
# ECMWF data skim asserts
self.assertTrue(os.path.isdir("output/ECMWF_ERA-5"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-5/tco3_zm.nc"))
self.assertTrue(os.path.isdir("output/ECMWF_ERA-i"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-i/tco3_zm.nc"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-i/vmro3_zm.nc"))
# Checks the original data has not been modified
self.assert_with_backup()
# Removes output data for other tests
self.clean_output()
def test_003_OutputSplitByYear(self):
"""Skims the data into the output folder spliting by year"""
with o3skim.utils.cd("data"):
ds = {name: o3skim.Source(name, collection) for
name, collection in self.config_base.items()}
with o3skim.utils.cd("output"):
[source.skim(groupby="year") for source in ds.values()]
# CCMI-1 data skim asserts
self.assertTrue(os.path.isdir("output/CCMI-1_IPSL"))
self.assertTrue(os.path.exists("output/CCMI-1_IPSL/tco3_zm_2000-2001.nc"))
self.assertTrue(os.path.exists("output/CCMI-1_IPSL/vmro3_zm_2000-2001.nc"))
# ECMWF data skim asserts
self.assertTrue(os.path.isdir("output/ECMWF_ERA-5"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-5/tco3_zm_2000-2001.nc"))
self.assertTrue(os.path.isdir("output/ECMWF_ERA-i"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-i/tco3_zm_2000-2001.nc"))
self.assertTrue(os.path.exists("output/ECMWF_ERA-i/vmro3_zm_2000-2001.nc"))
# Checks the original data has not been modified
self.assert_with_backup()
# Removes output data for other tests
self.clean_output()
def test_004_SourceErrorDontBreak(self):
"""The execution does not stop by an error in source"""
with o3skim.utils.cd("data"):
ds = {name: o3skim.Source(name, collection) for
name, collection in self.config_err.items()}
with o3skim.utils.cd("output"):
[source.skim() for source in ds.values()]
# ECMWF data skim asserts
self.assertTrue(os.path.isdir("output/ErrorModels_correct_variable"))
self.assertTrue(os.path.exists( "output/ErrorModels_correct_variable/vmro3_zm.nc"))
self.assertTrue(os.path.isdir( "output/ErrorModels_non_existing_variable"))
self.assertTrue( len(os.listdir("output/ErrorModels_non_existing_variable")) == 0)
# self.assertTrue(os.path.isdir("output/ECMWF_wrong_coordinates"))
# self.assertTrue(len(os.listdir("output/ECMWF_wrong_coordinates")) == 0)
# Checks the original data has not been modified
self.assert_with_backup()
# Removes output data for other tests
self.clean_output()
import xarray
import tests.mockup_data as mockup_data
# Test configurations ----------------------------------------------
configuration_file = "tests/sources_example.yaml"
year_line = range(2000, 2022)