Commit b1fe6955 authored by BorjaEst's avatar BorjaEst
Browse files

Edit functional tests to split by action

parent 0b1bf15a
import unittest
import pandas as pd
import numpy as np
import xarray as xr
from o3skim import source
model = xr.Dataset(
data_vars=dict(
tco3_zm=(["lon", "lat", "time"], np.random.rand(3, 3, 25)),
vmro3_zm=(["lon", "lat", "plev", "time"], np.random.rand(3, 3, 4, 25))
),
coords=dict(
lon=[-180, 0, 180],
lat=[-90, 0, 90],
plev=[1, 10, 100, 1000],
time=pd.date_range("2000-01-01", periods=25, freq='A')
),
attrs=dict(description="Test dataset")
)
class TestsModel(unittest.TestCase):
def assertHasAttr(self, obj, intendedAttr):
testBool = hasattr(obj, intendedAttr)
msg = 'obj lacking an attribute. obj: %s, intendedAttr: %s' % (
obj, intendedAttr)
self.assertTrue(testBool, msg=msg)
def test_dataset_has_model_accessor(self):
self.assertHasAttr(model, 'model')
import unittest
import pandas as pd
import numpy as np
import xarray as xr
from o3skim import source
Source = source.Source
name = "SourceTest"
collections = {} # Empty, only to test constructor stability
class TestsSource(unittest.TestCase):
def setUp(self):
self.source = Source(name, collections)
def test_property_name(self):
self.assertEqual(name, self.source.name)
def test_property_models(self):
expected = list(collections.keys())
result = self.source.models
self.assertEqual(expected, result)
......@@ -6,28 +6,52 @@ import o3skim
import pytest
import tests.mockup as mockup_data
import xarray
import yaml
# configurations ----------------------------------------------------
year_line = range(2000, 2022)
sources = ["SourceSplit", "SourceMerged"]
# Source files
sources_example = "tests/sources_example.yaml"
sources_err = "tests/sources_err.yaml"
available_configurations = {
'correct': {sources_example},
'with_errors': {sources_err}
}
models = ["ModelTCO3", "ModelVMRO3", "ModelALL"]
models_tco3 = ['ModelTCO3', 'ModelALL']
models_vmro3 = ['ModelVMRO3', 'ModelALL']
# Sources
sources = ["SourceSplit", "SourceMerged"]
# Models
tco3_models = {
'SourceSplit_ModelTCO3',
'SourceSplit_ModelALL',
'SourceMerged_ModelTCO3',
'SourceMerged_ModelALL'
}
vmro3_models = {
'SourceSplit_ModelVMRO3',
'SourceSplit_ModelALL',
'SourceMerged_ModelVMRO3',
'SourceMerged_ModelALL'
}
available_models = {
'all': tco3_models | vmro3_models,
'with_tco3': tco3_models,
'only_tco3': tco3_models - vmro3_models,
'with_vmro3': vmro3_models,
'only_vmro3': vmro3_models - tco3_models
}
# session fixtures ---------------------------------------------------
@pytest.fixture(scope='session')
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.cd(source_dir):
with o3skim.utils.cd(source_dir):
if source == 'SourceMerged':
mockup_data.combined(year_line)
mockup_data.noise(name='merged_noise.nc')
......@@ -40,99 +64,153 @@ def data_dir(tmpdir_factory):
@pytest.fixture(scope='session')
def config_file(request):
def output_dir(tmpdir_factory, groupby, model):
output_dir = tmpdir_factory.mktemp(model + "_by_" + str(groupby))
return output_dir
@pytest.fixture(scope='session')
def metadata_dir(tmpdir_factory, model):
metadata_dir = tmpdir_factory.mktemp(model + "_metadata")
return metadata_dir
@pytest.fixture(scope='session')
def config(request):
return request.param
@pytest.fixture(scope='session')
def config_dict(config_file):
return o3skim.load(config_file)
def model(request):
return request.param
# package fixtures --------------------------------------------------
@pytest.fixture(scope='session')
def config_dict(config):
return o3skim.utils.load(config)
# module fixtures ---------------------------------------------------
@pytest.fixture(scope='module')
def output_dir(tmpdir_factory):
return tmpdir_factory.mktemp("output")
@pytest.fixture(scope='session')
def load_model(data_dir, config_dict, model):
model_configuration = config_dict[model]
with o3skim.utils.cd(data_dir):
return o3skim.loading(**model_configuration)
@pytest.fixture(scope='module')
def groupby(request):
@pytest.fixture(scope='session')
def dataset(load_model):
dataset, metadata = load_model
return dataset
@pytest.fixture(scope='session')
def metadata(load_model):
dataset, metadata = load_model
return metadata
@pytest.fixture(scope='session')
def actions(request):
return request.param
@pytest.fixture(scope='module')
def source_name(request):
@pytest.fixture(scope='session')
def processed(dataset, actions):
return o3skim.processing(dataset, actions)
@pytest.fixture(scope='session')
def groupby(request):
return request.param
@pytest.fixture(scope='module')
def source(config_dict, source_name, data_dir):
with o3skim.cd(data_dir):
source = o3skim.Source(source_name, **config_dict[source_name])
return source
@pytest.fixture(scope='session')
def splitted(dataset, groupby):
return o3skim.group(dataset, split_by=groupby)
@pytest.fixture(scope='session')
def years(splitted):
years, splitted_ds = splitted
return years
@pytest.fixture(scope='module')
def skimmed(groupby, source, output_dir):
with o3skim.cd(output_dir):
source.skim(groupby=groupby)
yield groupby, source.name
@pytest.fixture(scope='session')
def splitted_ds(splitted):
years, splitted_ds = splitted
return splitted_ds
# class fixtures --------------------------------------------------
@pytest.fixture(scope='session')
def variables(model):
variables = []
if model in available_models['with_tco3']:
variables.append('tco3_zm')
if model in available_models['with_vmro3']:
variables.append('vmro3_zm')
return variables
# function fixtures -------------------------------------------------
@pytest.fixture()
def model_name(request):
return request.param
@pytest.fixture(scope='session')
def expected_netCDF(groupby, variables):
if groupby == None:
return ["{}.nc".format(v)
for v in variables]
elif groupby == 'year':
def format(v, y): return "{}_{}-{}.nc".format(v, y, y+1)
return [format(v, y)
for y in year_line
for v in variables]
elif groupby == 'decade':
def format(v, y): return "{}_{}-{}.nc".format(v, y, y+10)
return [format(v, y)
for y in year_line if y % 10 == 0
for v in variables]
else:
raise KeyError("bad groupby {}".format(groupby))
@pytest.fixture()
def model(source, model_name):
return source[model_name]
@pytest.fixture(scope='session')
def save_netCDF(output_dir, splitted_ds, groupby, years):
with o3skim.utils.cd(output_dir):
o3skim.saving(splitted_ds, groupby, years)
@pytest.fixture()
def year(request):
return request.param
@pytest.fixture(scope='session')
def saved_ds(output_dir, save_netCDF, expected_netCDF):
with o3skim.utils.cd(output_dir):
return xarray.open_mfdataset(expected_netCDF)
@pytest.fixture()
def variable(request):
return request.param
@pytest.fixture(scope='session')
def save_metadata(metadata_dir, metadata):
with o3skim.utils.cd(metadata_dir):
o3skim.utils.save("metadata.yaml", metadata)
@pytest.fixture()
def metadata_file(skimmed, model_name):
_, source_name = skimmed
with o3skim.cd("{}_{}".format(source_name, model_name)):
yield "metadata.yaml"
@pytest.fixture(scope='session')
def metadata_file_content(metadata_dir, save_metadata):
with o3skim.utils.cd(metadata_dir):
return o3skim.utils.load("metadata.yaml")
@pytest.fixture()
def metadata_dict(metadata_file):
with open(metadata_file, "r") as ymlfile:
config = yaml.safe_load(ymlfile)
return config
# package fixtures --------------------------------------------------
# module fixtures ---------------------------------------------------
# class fixtures --------------------------------------------------
# function fixtures -------------------------------------------------
@pytest.fixture()
def skimed_file(skimmed, model_name, variable, year):
groupby, source_name = skimmed
with o3skim.cd("{}_{}".format(source_name, model_name)):
if not groupby:
yield "{}.nc".format(variable)
if groupby == 'year':
yield "{}_{}-{}.nc".format(variable, year, year + 1)
if groupby == 'decade':
yield "{}_{}-{}.nc".format(variable, year, year + 10)
def output_cd(output_dir):
with o3skim.utils.cd(output_dir):
yield None
@pytest.fixture()
def skimmed_model(skimed_file):
with xarray.open_dataset(skimed_file) as model:
yield model
def metadata_cd(metadata_dir):
with o3skim.utils.cd(metadata_dir):
yield None
SourceSplit:
metadata:
meta_0: Source metadata string example
meta_1: Source metadata to be replaced by Model
meta_2:
meta_20: Sub-metadata from Source
meta_21: Sub-metadata to be replaced by Model
ModelTCO3:
SourceSplit_ModelTCO3:
metadata:
meta_1: Model metadata string example
meta_2:
......@@ -20,7 +13,7 @@ SourceSplit:
metadata:
meta_tco3_1: TCO3 metadata string example
meta_tco3_2: 0
ModelVMRO3:
SourceSplit_ModelVMRO3:
metadata:
meta_1: Model metadata string example
meta_2:
......@@ -36,7 +29,7 @@ SourceSplit:
metadata:
meta_vmro3_1: VMRO3 metadata string example
meta_vmro3_2: 0
ModelALL:
SourceSplit_ModelALL:
metadata:
meta_1: Model metadata string example
meta_2:
......@@ -62,14 +55,7 @@ SourceSplit:
plev: pressure_level
lat: latitude
lon: longitude
SourceMerged:
metadata:
meta_0: Source metadata string example
meta_1: Source metadata to be replaced by Model
meta_2:
meta_20: Sub-metadata from Source
meta_21: Sub-metadata to be replaced by Model
ModelTCO3:
SourceMerged_ModelTCO3:
metadata:
meta_1: Model metadata string example
meta_2:
......@@ -84,7 +70,7 @@ SourceMerged:
time: time
lat: latitude
lon: longitude
ModelVMRO3:
SourceMerged_ModelVMRO3:
metadata:
meta_1: Model metadata string example
meta_2:
......@@ -100,7 +86,7 @@ SourceMerged:
plev: pressure_level
lat: latitude
lon: longitude
ModelALL:
SourceMerged_ModelALL:
metadata:
meta_1: Model metadata string example
meta_2:
......
......@@ -6,16 +6,30 @@ configs = conftest.available_configurations
models = conftest.available_models
@pytest.mark.parametrize('config', configs['correct'], indirect=True)
@pytest.mark.parametrize('model', models['all'], indirect=True)
class TestO3CommonsLoad:
def test_attrs(self, dataset):
assert dataset.attrs == dict(description="Test ozone dataset")
def test_metadata(self, metadata):
assert metadata["meta_1"] == "Model metadata string example"
assert metadata["meta_2"]["meta_21"] == "Sub-metadata from Model"
@pytest.mark.parametrize('config', configs['correct'], indirect=True)
class TestTCO3Load:
@pytest.mark.parametrize('model', models['with_tco3'], indirect=True)
def test_variable_load(self, dataset):
assert 'tco3_zm' in dataset
assert 'time' in dataset['tco3_zm'].coords
assert 'lat' in dataset['tco3_zm'].coords
assert 'lon' in dataset['tco3_zm'].coords
assert len(dataset['tco3_zm'].coords) == 3
darray = dataset['tco3_zm']
assert 'time' in darray.coords
assert 'lat' in darray.coords
assert 'lon' in darray.coords
assert len(darray.coords) == 3
assert darray.attrs == dict(description="Test tco3 xarray")
@pytest.mark.parametrize('model', models['with_tco3'], indirect=True)
def test_coordinates(self, dataset):
......@@ -34,11 +48,13 @@ class TestVMRO3Load:
@pytest.mark.parametrize('model', models['with_vmro3'], indirect=True)
def test_variable_load(self, dataset):
assert 'vmro3_zm' in dataset
assert 'time' in dataset['vmro3_zm'].coords
assert 'plev' in dataset['vmro3_zm'].coords
assert 'lat' in dataset['vmro3_zm'].coords
assert 'lon' in dataset['vmro3_zm'].coords
assert len(dataset['vmro3_zm'].coords) == 4
darray = dataset['vmro3_zm']
assert 'time' in darray.coords
assert 'plev' in darray.coords
assert 'lat' in darray.coords
assert 'lon' in darray.coords
assert len(darray.coords) == 4
assert darray.attrs == dict(description="Test vmro3 xarray")
@pytest.mark.parametrize('model', models['with_vmro3'], indirect=True)
def test_coordinates(self, dataset):
......
......@@ -9,11 +9,23 @@ models = conftest.available_models
operations = ['lon_mean', 'lat_mean']
actions = list(itertools.permutations(operations))
using = {
'any': [list(x) for x in actions],
'lon_mean': [list(x) for x in actions if 'lon_mean' in x],
'lat_mean': [list(x) for x in actions if 'lat_mean' in x]
}
@pytest.mark.parametrize('config', configs['correct'], indirect=True)
@pytest.mark.parametrize('model', models['all'], indirect=True)
@pytest.mark.parametrize('actions', using['any'], indirect=True)
class TestCommon:
def test_attrs(self, dataset, processed):
assert dataset.attrs == processed.attrs
for o3var in processed.var():
assert dataset[o3var].attrs == processed[o3var].attrs
@pytest.mark.parametrize('config', configs['correct'], indirect=True)
@pytest.mark.parametrize('model', models['all'], indirect=True)
class TestLonMean:
......
......@@ -15,7 +15,7 @@ models = conftest.available_models
class TestNetCDFSaving:
def test_arefiles(self, output_cd, save_netCDF, expected_netCDF):
assert all([os.path.isfile(f) for f in expected_netCDF])
assert set(os.listdir()) == set(expected_netCDF)
def test_saved_data(self, dataset, saved_ds, variables):
xr.testing.assert_equal(dataset, saved_ds)
......@@ -28,9 +28,9 @@ class TestNetCDFSaving:
@pytest.mark.parametrize('config', configs['correct'], indirect=True)
@pytest.mark.parametrize('model', models['all'], indirect=True)
class TestMetadataFSaving:
class TestMetadataSaving:
def test_isfile(self, output_cd, save_metadata):
def test_isfile(self, metadata_cd, save_metadata):
assert os.path.isfile("metadata.yaml")
def test_content(self, metadata, metadata_file_content):
......
"""Pytest module to test sources as blackbox."""
import os
import o3skim
import pytest
import tests.conftest as conftest
import xarray
@pytest.mark.parametrize('config_file', [conftest.sources_example], indirect=True)
@pytest.mark.parametrize('source_name', conftest.sources, indirect=True)
class TestSource:
def test_constructor(self, config_dict, source_name, data_dir):
with o3skim.cd(data_dir):
source = o3skim.Source(source_name, **config_dict[source_name])
assert type(source.name) is str
assert type(source.models) is list
assert source.models != []
def test_skimming(self, source, output_dir):
with o3skim.cd(output_dir):
assert None == source.skim(groupby=None)
assert None == source.skim(groupby='year')
assert None == source.skim(groupby='decade')
@pytest.mark.parametrize('model_name', conftest.models, indirect=True)
def test_model_properties(self, model):
assert isinstance(model, xarray.Dataset)
assert model.time.size != 0
@pytest.mark.parametrize('config_file', [conftest.sources_err], indirect=True)
@pytest.mark.parametrize('source_name', ["SourceSplit"], indirect=True)
class TestExceptions:
def test_constructor(self, config_dict, source_name, data_dir):
with o3skim.cd(data_dir):
source = o3skim.Source(source_name, config_dict[source_name])
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
"""Pytest module to test tco3 data as blackbox."""
import os
import o3skim
import pytest
import tests.conftest as conftest
import xarray
def pytest_generate_tests(metafunc):
config_file = conftest.sources_example
metafunc.parametrize("config_file", [config_file], indirect=True)
metafunc.parametrize("source_name", conftest.sources, indirect=True)
metafunc.parametrize("variable", ["tco3_zm"], indirect=True)
class TestModel:
@pytest.mark.parametrize('model_name', conftest.models_tco3, indirect=True)
def test_tco3_coordinates(self, model, variable):
assert 'time' in model.coords
assert 'lat' in model.coords
assert 'lon' in model.coords
@pytest.mark.parametrize('model_name', ["ModelTCO3"], indirect=True)
def test_tco3_no_extra_coordinates(self, model, variable):
assert len(model.coords) == 3
@pytest.mark.parametrize('groupby', [None], indirect=True)
class TestSkimming_NoGroup:
@pytest.mark.parametrize('model_name', conftest.models_tco3, indirect=True)
@pytest.mark.parametrize('year', [None], indirect=True)
def test_isTCO3file(self, skimed_file):
assert os.path.isfile(skimed_file)
@pytest.mark.parametrize('model_name', conftest.models_tco3, indirect=True)
@pytest.mark.parametrize('year', [None], indirect=True)
def test_skimmed_tco3_coords(self, skimmed_model):
assert 'tco3_zm' in skimmed_model.var()
assert 'time' in skimmed_model.coords
assert not 'plev' in skimmed_model.coords
assert 'lat' in skimmed_model.coords
assert not 'lon' in skimmed_model.coords
@pytest.mark.parametrize('model_name', conftest.models_tco3, indirect=True)
@pytest.mark.parametrize('year', [None], indirect=True)
def test_skimmed_tco3_attrs(self, skimmed_model):
expected_ds_attrs = dict(description="Test ozone dataset")
assert skimmed_model.attrs == expected_ds_attrs
expected_tco3_attrs = dict(description="Test tco3 xarray")
assert skimmed_model.tco3_zm.attrs == expected_tco3_attrs
@pytest.mark.parametrize('groupby', ['year'], indirect=True)
class TestSkimming_ByYear:
@pytest.mark.parametrize('model_name', conftest.models_tco3, indirect=True)
@pytest.mark.parametrize('year', list(conftest.year_line), indirect=True)
def test_isTCO3file(self, skimed_file):
assert os.path.isfile(skimed_file)