Commit d18d90dd authored by Javier Vegas-Regidor's avatar Javier Vegas-Regidor
Browse files

Added missing tests for ocean and statistics diagnostics

parent db2c5699
......@@ -51,8 +51,8 @@ class AverageSection(Diagnostic):
self.domain == other.domain and self.variable == other.variable and self.box == other.box
def __str__(self):
return 'Average section Startdate: {0} Member: {1} Chunk: {2} Box: {3} ' \
'Variable: {4}:{5}'.format(self.startdate, self.member, self.chunk, self.box, self.domain, self.variable)
return 'Average section Startdate: {0.startdate} Member: {0.member} Chunk: {0.chunk} Box: {0.box} ' \
'Variable: {0.domain}:{0.variable} Grid: {0.grid}'.format(self)
@classmethod
def generate_jobs(cls, diags, options):
......
......@@ -48,17 +48,14 @@ class HeatContent(Diagnostic):
self.box = box
self.min_level = min_level
self.max_level = max_level
self.required_vars = ['so', 'mlotst']
self.generated_vars = ['scvertsum']
def __eq__(self, other):
return self.startdate == other.startdate and self.member == other.member and self.chunk == other.chunk and \
self.box == other.box and self.basin == other.basin and self.mxloption == other.mxloption
def __str__(self):
return 'Heat content Startdate: {0} Member: {1} Chunk: {2} Mixed layer: {3} Box: {4} ' \
'Basin: {5}'.format(self.startdate, self.member, self.chunk, self.mxloption, self.box,
self.basin.fullname)
return 'Heat content Startdate: {0.startdate} Member: {0.member} Chunk: {0.chunk} Mixed layer: {0.mxloption} ' \
'Box: {0.box} Basin: {0.basin}'.format(self)
@classmethod
def generate_jobs(cls, diags, options):
......
......@@ -91,11 +91,12 @@ class Interpolate(Diagnostic):
options = cls.process_options(options, options_available)
job_list = list()
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(
Interpolate(diags.data_manager, startdate, member, chunk,
options['domain'], options['variable'], options['target_grid'],
diags.config.experiment.model_version, options['invert_lat'], options['original_grid']))
for var in options['variable']:
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(
Interpolate(diags.data_manager, startdate, member, chunk,
options['domain'], var , options['target_grid'],
diags.config.experiment.model_version, options['invert_lat'], options['original_grid']))
return job_list
def request_data(self):
......
......@@ -62,16 +62,16 @@ class InterpolateCDO(Diagnostic):
def __eq__(self, other):
return self.startdate == other.startdate and self.member == other.member and self.chunk == other.chunk and \
self.model_version == other.model_version and self.domain == other.domain and \
self.variable == other.variable and self.grid == other.grid and self.original_grid == other.original_grid
self.variable == other.variable and self.mask_oceans == other.mask_oceans and self.grid == other.grid and \
self.original_grid == other.original_grid
def __str__(self):
return 'Interpolate with CDO Startdate: {0} Member: {1} Chunk: {2} ' \
'Variable: {3}:{4} Target grid: {5} ' \
'Model: {6}' .format(self.startdate, self.member, self.chunk, self.domain, self.variable, self.grid,
self.model_version)
return 'Interpolate with CDO Startdate: {0.startdate} Member: {0.member} Chunk: {0.chunk} ' \
'Variable: {0.domain}:{0.variable} Target grid: {0.grid} Original grid: {0.original_grid} ' \
'Mask ocean: {0.mask_oceans} Model: {0.model_version}'.format(self)
@classmethod
def generate_jobs(cls, diags, options):
def generate_jobs(cls, diags, options ):
"""
Creates a job for each chunk to compute the diagnostic
......@@ -82,10 +82,9 @@ class InterpolateCDO(Diagnostic):
:return:
"""
options_available = (DiagnosticDomainOption(default_value=ModelingRealms.ocean),
DiagnosticVariableOption(),
DiagnosticVariableListOption('variables'),
DiagnosticOption('target_grid', diags.config.experiment.atmos_grid.lower()),
DiagnosticChoiceOption('method', InterpolateCDO.METHODS, InterpolateCDO.BILINEAR),
DiagnosticChoiceOption('method', InterpolateCDO.METHODS, InterpolateCDO.BILINEAR),
DiagnosticBoolOption('mask_oceans', True),
DiagnosticOption('original_grid', ''),
DiagnosticBoolOption('weights_from_mask', True)
......@@ -97,6 +96,17 @@ class InterpolateCDO(Diagnostic):
job_list = list()
weights = TempFile.get()
method = options['method'].lower()
cls._compute_weights(diags, method, options, target_grid, weights)
for var in options['variables']:
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(InterpolateCDO(diags.data_manager, startdate, member, chunk,
options['domain'], var, target_grid,
diags.config.experiment.model_version, options['mask_oceans'],
options['original_grid'], weights))
return job_list
@classmethod
def _compute_weights(cls, diags, method, options, target_grid, weights):
if options['weights_from_mask']:
temp = cls.get_sample_grid_file()
else:
......@@ -113,13 +123,6 @@ class InterpolateCDO(Diagnostic):
Utils.cdo.gencon2(target_grid, input=temp, output=weights)
os.remove(temp)
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(InterpolateCDO(diags.data_manager, startdate, member, chunk,
options['domain'], options['variable'], target_grid,
diags.config.experiment.model_version, options['mask_oceans'],
options['original_grid'], weights))
return job_list
@classmethod
def get_sample_grid_file(cls):
temp = TempFile.get()
......
# coding=utf-8
from earthdiagnostics.diagnostic import Diagnostic, DiagnosticVariableOption, \
from earthdiagnostics.diagnostic import Diagnostic, DiagnosticVariableListOption, \
DiagnosticDomainOption, DiagnosticChoiceOption, DiagnosticOption
from earthdiagnostics.utils import Utils, TempFile
import numpy as np
......@@ -56,25 +56,33 @@ class MaskLand(Diagnostic):
:return:
"""
options_available = (DiagnosticDomainOption('domain'),
DiagnosticVariableOption('variable'),
DiagnosticVariableListOption('variables'),
DiagnosticChoiceOption('cell', ('t', 'u', 'v', 'f', 'w'), 't'),
DiagnosticOption('grid', ''))
options = cls.process_options(options, options_available)
mask_file = Utils.openCdf('mask.nc')
cell_point = options['cell']
# W and T share the same mask
if cell_point == 'w':
cell_point = 't'
mask = mask_file.variables['{0}mask'.format(cell_point)][:].astype(float)
mask[mask == 0] = np.nan
mask = cls._get_mask(cell_point)
job_list = list()
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(MaskLand(diags.data_manager, startdate, member, chunk,
options['domain'], options['variable'], mask, options['grid']))
for var in options['variables']:
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(MaskLand(diags.data_manager, startdate, member, chunk,
options['domain'], var, mask, options['grid']))
return job_list
@classmethod
def _get_mask(cls, cell_point):
mask_file = Utils.openCdf('mask.nc')
mask = mask_file.variables['{0}mask'.format(cell_point)][:].astype(float)
mask[mask == 0] = np.nan
mask_file.close()
return mask
"Diagnostic alias for the configuration file"
def request_data(self):
......
......@@ -58,8 +58,9 @@ class RegionMean(Diagnostic):
self.box == other.box and self.variable == other.variable
def __str__(self):
return 'Region mean Startdate: {0} Member: {1} Chunk: {2} Variable: {3} ' \
'Box: {4}'.format(self.startdate, self.member, self.chunk, self.variable, self.box)
return 'Region mean Startdate: {0.startdate} Member: {0.member} Chunk: {0.chunk} Variable: {0.variable} ' \
'Grid point: {0.grid_point} Box: {0.box} Save 3D: {0.save3d} Save variance: {0.variance} ' \
'Original grid: {0.grid}'.format(self)
@classmethod
def generate_jobs(cls, diags, options):
......
......@@ -45,12 +45,13 @@ class DaysOverPercentile(Diagnostic):
self.startdate = '{0}{1:02}01'.format(self.year_to_compute, self.forecast_month)
def __eq__(self, other):
return self.startdate == other.startdate and self.domain == other.domain and self.variable == other.variable
return self.startdate == other.startdate and self.domain == other.domain and \
self.variable == other.variable and self.start_year == other.start_year and \
self.end_year == other.end_year
def __str__(self):
return 'Days over percentile Startdate: {0} ' \
'Variable: {1}:{2} Climatology: {3}-{4}'.format(self.startdate, self.domain, self.variable,
self.start_year, self.end_year)
return 'Days over percentile Startdate: {0.startdate} Variable: {0.domain}:{0.variable} ' \
'Climatology: {0.start_year}-{0.end_year}'.format(self)
@classmethod
def generate_jobs(cls, diags, options):
......
......@@ -11,7 +11,7 @@ source_path = os.path.join(work_path, '..', 'earthdiagnostics', '*')
print(source_path)
cov = coverage.Coverage(include=source_path)
cov.set_option("run:branch", True)
cov.set_option("html:title", 'Coverage report for ESMValTool')
cov.set_option("html:title", 'Coverage report for EarthDiagnostics')
cov.start()
suite = unittest.TestLoader().discover('.')
......
# # coding=utf-8
# from unittest import TestCase
#
# from earthdiagnostics.box import Box
# from earthdiagnostics.ocean.averagesection import AverageSection
# from mock import Mock
#
# from earthdiagnostics.modelingrealm import ModelingRealms
#
#
# class TestAverageSection(TestCase):
#
# def setUp(self):
# self.data_manager = Mock()
# self.diags = Mock()
#
# self.box = Box()
# self.box.min_lat = 0
# self.box.max_lat = 0
# self.box.min_lon = 0
# self.box.max_lon = 0
#
# self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
# self.psi = AverageSection(self.data_manager, '20000101', 1, 1, ModelingRealms.ocean, 'var', self.box)
#
# def test_generate_jobs(self):
# jobs = AverageSection.generate_jobs(self.diags, ['diagnostic', 'var', '0', '0', '0', '0'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], AverageSection(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var',
# self.box))
# self.assertEqual(jobs[1], AverageSection(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var',
# self.box))
#
# jobs = AverageSection.generate_jobs(self.diags, ['diagnostic', 'var', '0', '0', '0', '0', 'ocean'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], AverageSection(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var',
# self.box))
# self.assertEqual(jobs[1], AverageSection(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var',
# self.box))
#
# with self.assertRaises(Exception):
# AverageSection.generate_jobs(self.diags, ['diagnostic'])
# with self.assertRaises(Exception):
# AverageSection.generate_jobs(self.diags, ['diagnostic', '0', '0', '0', '0', '0', '0', '0'])
#
# def test_str(self):
# self.assertEquals(str(self.psi), 'Average section Startdate: 20000101 Member: 1 Chunk: 1 Box: 0N0E '
# 'Variable: ocean:var')
# coding=utf-8
from unittest import TestCase
from earthdiagnostics.box import Box
from earthdiagnostics.ocean.averagesection import AverageSection
from mock import Mock, patch
from earthdiagnostics.diagnostic import DiagnosticVariableOption
from earthdiagnostics.modelingrealm import ModelingRealms
class TestAverageSection(TestCase):
def setUp(self):
self.data_manager = Mock()
self.diags = Mock()
self.box = Box()
self.box.min_lat = 0
self.box.max_lat = 0
self.box.min_lon = 0
self.box.max_lon = 0
self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
def fake_parse(self, value):
return value
@patch.object(DiagnosticVariableOption, 'parse', fake_parse)
def test_generate_jobs(self):
jobs = AverageSection.generate_jobs(self.diags, ['diagnostic', 'ocean', 'var', '0', '0', '0', '0'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], AverageSection(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var',
self.box, ''))
self.assertEqual(jobs[1], AverageSection(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var',
self.box, ''))
jobs = AverageSection.generate_jobs(self.diags, ['diagnostic', 'ocean', 'var', '0', '0', '0', '0', 'grid'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], AverageSection(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var',
self.box, 'grid'))
self.assertEqual(jobs[1], AverageSection(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var',
self.box, 'grid'))
with self.assertRaises(Exception):
AverageSection.generate_jobs(self.diags, ['diagnostic'])
with self.assertRaises(Exception):
AverageSection.generate_jobs(self.diags, ['diagnostic', 'ocean', 'var', '0', '0', '0', '0', 'grid',
'extra'])
def test_str(self):
diag = AverageSection(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var', self.box, 'grid')
self.assertEquals(str(diag), 'Average section Startdate: 20010101 Member: 0 Chunk: 0 Box: 0N0E '
'Variable: ocean:var Grid: grid')
# # coding=utf-8
# from unittest import TestCase
#
# from earthdiagnostics.box import Box
# from earthdiagnostics.constants import Basins
# from earthdiagnostics.ocean.heatcontent import HeatContent
# from mock import Mock
#
#
# class TestHeatContent(TestCase):
#
# def setUp(self):
# self.data_manager = Mock()
#
# self.diags = Mock()
# self.diags.model_version = 'model_version'
# self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
#
# self.box = Box(False)
# self.box.min_depth = 0
# self.box.max_depth = 100
#
# self.heat_content = HeatContent(self.data_manager, '20000101', 1, 1, Basins().Global, 1, self.box)
#
# def test_generate_jobs(self):
# jobs = HeatContent.generate_jobs(self.diags, ['diagnostic', 'atl', '-1', '0', '100'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], HeatContent(self.data_manager, '20010101', 0, 0, Basins().Atlantic, -1, self.box))
# self.assertEqual(jobs[1], HeatContent(self.data_manager, '20010101', 0, 1, Basins().Atlantic, -1, self.box))
#
# with self.assertRaises(Exception):
# HeatContent.generate_jobs(self.diags, ['diagnostic'])
#
# with self.assertRaises(Exception):
# HeatContent.generate_jobs(self.diags, ['diagnostic', '0', '0', '0', '0', '0', '0', '0'])
#
# def test_str(self):
# self.assertEquals(str(self.heat_content), 'Heat content Startdate: 20000101 Member: 1 Chunk: 1 Mixed layer: 1 '
# 'Box: 0-100 Basin: Global_Ocean')
# coding=utf-8
from unittest import TestCase
from earthdiagnostics.box import Box
from earthdiagnostics.constants import Basins
from earthdiagnostics.ocean.heatcontent import HeatContent
from mock import Mock, patch
def _get_levels_from_meters_mock(cls, box):
return 20, 10
class TestHeatContent(TestCase):
def setUp(self):
self.data_manager = Mock()
self.diags = Mock()
self.diags.model_version = 'model_version'
self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
self.box = Box(True)
self.box.min_depth = 0
self.box.max_depth = 100
@patch('earthdiagnostics.ocean.heatcontent.HeatContent._get_levels_from_meters')
def test_generate_jobs(self, levels_mock):
levels_mock.return_value = (1, 20)
jobs = HeatContent.generate_jobs(self.diags, ['diagnostic', 'Global', '-1', '0', '100'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], HeatContent(self.data_manager, '20010101', 0, 0, Basins().Global, -1,
self.box, 0, 0))
self.assertEqual(jobs[1], HeatContent(self.data_manager, '20010101', 0, 1, Basins().Global, -1,
self.box, 0, 0))
with self.assertRaises(Exception):
HeatContent.generate_jobs(self.diags, ['diagnostic'])
with self.assertRaises(Exception):
HeatContent.generate_jobs(self.diags, ['diagnostic', '0', '0', '0', '0', '0', '0', '0'])
def test_str(self):
diag = HeatContent(self.data_manager, '20010101', 0, 0, Basins().Global, -1, self.box, 1, 20)
self.assertEquals(str(diag), 'Heat content Startdate: 20010101 Member: 0 Chunk: 0 Mixed layer: -1 Box: 0-100m '
'Basin: Global')
......@@ -20,7 +20,7 @@ class TestHeatContentLayer(TestCase):
self.box.min_depth = 0
self.box.max_depth = 100
self.psi = HeatContentLayer(self.data_manager, '20000101', 1, 1, self.box, self.weight, 0, 10)
def test_str(self):
self.assertEquals(str(self.psi), 'Heat content layer Startdate: 20000101 Member: 1 Chunk: 1 Box: 0-100m')
diag = HeatContentLayer(self.data_manager, '20000101', 1, 1, self.box, self.weight, 0, 10)
self.assertEquals(str(diag), 'Heat content layer Startdate: 20000101 Member: 1 Chunk: 1 Box: 0-100m')
# # coding=utf-8
# from unittest import TestCase
#
# from earthdiagnostics.ocean.interpolate import Interpolate
# from mock import Mock
#
# from earthdiagnostics.modelingrealm import ModelingRealms
#
#
# class TestInterpolate(TestCase):
#
# def setUp(self):
# self.data_manager = Mock()
#
# self.diags = Mock()
# self.diags.model_version = 'model_version'
# self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
# self.diags.config.experiment.model_version = 'model_version'
#
# self.interpolate = Interpolate(self.data_manager, '20000101', 1, 1, ModelingRealms.atmos, 'var', 'grid',
# 'model_version', False)
#
# def test_generate_jobs(self):
# jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var', 'grid',
# 'model_version', False))
# self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var', 'grid',
# 'model_version', False))
#
# jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
# 'model_version', False))
# self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.atmos, 'var', 'grid',
# 'model_version', False))
#
# jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos', 'true'])
# self.assertEqual(len(jobs), 2)
# self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
# 'model_version', True))
# self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.atmos, 'var', 'grid',
# 'model_version', True))
#
# with self.assertRaises(Exception):
# Interpolate.generate_jobs(self.diags, ['interp'])
#
# with self.assertRaises(Exception):
# Interpolate.generate_jobs(self.diags, ['interp', '0', '0', '0', '0', '0', '0', '0'])
#
# def test_str(self):
# self.assertEquals(str(self.interpolate), 'Interpolate Startdate: 20000101 Member: 1 Chunk: 1 '
# 'Variable: atmos:var Target grid: grid Invert lat: False '
# 'Model: model_version')
# coding=utf-8
from unittest import TestCase
from earthdiagnostics.ocean.interpolate import Interpolate
from mock import Mock, patch
from earthdiagnostics.modelingrealm import ModelingRealms
from earthdiagnostics.diagnostic import DiagnosticVariableListOption
class TestInterpolate(TestCase):
def setUp(self):
self.data_manager = Mock()
self.diags = Mock()
self.diags.model_version = 'model_version'
self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
self.diags.config.experiment.model_version = 'model_version'
def fake_parse(self, value):
return value.split('-')
@patch.object(DiagnosticVariableListOption, 'parse', fake_parse)
def test_generate_jobs(self):
jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var', 'grid',
'model_version', False, ''))
self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var', 'grid',
'model_version', False, ''))
jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var1-var2'])
self.assertEqual(len(jobs), 4)
self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var1', 'grid',
'model_version', False, ''))
self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var1', 'grid',
'model_version', False, ''))
self.assertEqual(jobs[2], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.ocean, 'var2', 'grid',
'model_version', False, ''))
self.assertEqual(jobs[3], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.ocean, 'var2', 'grid',
'model_version', False, ''))
jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
'model_version', False, ''))
self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.atmos, 'var', 'grid',
'model_version', False, ''))
jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos', 'true'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
'model_version', True, ''))
self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.atmos, 'var', 'grid',
'model_version', True, ''))
jobs = Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos', 'true', 'original_grid'])
self.assertEqual(len(jobs), 2)
self.assertEqual(jobs[0], Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
'model_version', True, 'original_grid'))
self.assertEqual(jobs[1], Interpolate(self.data_manager, '20010101', 0, 1, ModelingRealms.atmos, 'var', 'grid',
'model_version', True, 'original_grid'))
with self.assertRaises(Exception):
Interpolate.generate_jobs(self.diags, ['interp'])
with self.assertRaises(Exception):
Interpolate.generate_jobs(self.diags, ['interp', 'grid', 'var', 'atmos', 'true', 'original_grid', 'extra'])
def test_str(self):
diag = Interpolate(self.data_manager, '20010101', 0, 0, ModelingRealms.atmos, 'var', 'grid',
'model_version', True, 'original_grid')
self.assertEquals(str(diag), 'Interpolate Startdate: 20010101 Member: 0 Chunk: 0 Variable: atmos:var '
'Target grid: grid Invert lat: True Model: model_version '
'Original grid: original_grid')
# coding=utf-8
from unittest import TestCase
from earthdiagnostics.ocean.interpolatecdo import InterpolateCDO
from mock import Mock, patch
from earthdiagnostics.modelingrealm import ModelingRealms
from earthdiagnostics.diagnostic import DiagnosticVariableListOption, DiagnosticOptionError
class TestInterpolate(TestCase):
def setUp(self):
self.data_manager = Mock()
self.diags = Mock()
self.diags.model_version = 'model_version'
self.diags.config.experiment.get_chunk_list.return_value = (('20010101', 0, 0), ('20010101', 0, 1))
self.diags.config.experiment.model_version = 'model_version'
self.diags.config.experiment.atmos_grid = 'atmos_grid'
def fake_parse(self, value):
if not value:
raise DiagnosticOptionError
return value.split('-')
@patch('earthdiagnostics.ocean.interpolatecdo.InterpolateCDO._compute_weights')
@patch.object(DiagnosticVariableListOption, 'parse', fake_parse)
def test_generate_jobs(self, mock_weights):