99 lines
4.7 KiB
Python
99 lines
4.7 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
from scripts.web.config import load_config
|
|
from scripts.web.parameters import catalog, validate_spec
|
|
|
|
|
|
class ParameterTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.temp = tempfile.TemporaryDirectory()
|
|
cls.config = load_config()
|
|
cls.schema = catalog(cls.config)
|
|
filename = Path(cls.temp.name) / 'parameters.json'
|
|
filename.write_text(json.dumps(cls.schema))
|
|
cls.config['_paths']['catalog'] = str(filename)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.temp.cleanup()
|
|
|
|
def validate(self, **sections):
|
|
return validate_spec(sections, self.config)
|
|
|
|
def test_relative_paths_and_catalog_from_another_directory(self):
|
|
project = Path(__file__).resolve().parents[1]
|
|
config = copy.deepcopy(self.config)
|
|
config['storage']['root'] = 'tmp/path-test-data'
|
|
config['python'] = '.venv/bin/python'
|
|
filename = Path(self.temp.name) / 'relative-config.json'
|
|
filename.write_text(json.dumps(config))
|
|
loaded = load_config(filename)
|
|
self.assertEqual(loaded['_python'], str(project / config['python']))
|
|
self.assertEqual(loaded['_paths']['database'], str(project / config['storage']['root'] / config['storage']['database']))
|
|
subprocess.run([sys.executable, str(project / 'scripts/web/parameters.py'), '--config', str(filename), '--output', 'relative-catalog.json'], cwd=self.temp.name, check=True, timeout=15, capture_output=True)
|
|
result = json.loads((Path(self.temp.name) / 'relative-catalog.json').read_text())
|
|
self.assertEqual(result['source_hash'], self.schema['source_hash'])
|
|
|
|
def test_defaults_come_from_model_and_configuration(self):
|
|
result = self.validate()
|
|
self.assertTrue(result['ok'])
|
|
self.assertEqual(result['spec']['run'], self.config['run_defaults'])
|
|
self.assertEqual(result['spec']['model']['ncpus'], self.config['limits']['max_cpus'])
|
|
fields = {f['name']: f for f in self.schema['fields']}
|
|
self.assertEqual(fields['flux']['default'], 1367.0)
|
|
self.assertFalse(fields['landmap']['editable'])
|
|
self.assertEqual(fields['aerofile']['type'], 'string')
|
|
self.assertTrue(fields['aerofile']['choices'])
|
|
|
|
def test_paths_scripts_and_raw_namelists_fail_closed(self):
|
|
for key, value in {'landmap': '/tmp/a', 'topomap': '/tmp/a', 'runscript': 'x',
|
|
'resources': ['/tmp/a'], 'otherargs': {'X@a': '1'}, 'noutput': False,
|
|
'runsteps': 99}.items():
|
|
with self.subTest(key=key):
|
|
result = self.validate(parameters={key: value})
|
|
self.assertFalse(result['ok'])
|
|
self.assertIn('parameters.' + key, result['errors'])
|
|
|
|
def test_types_enumerations_finite_values_and_unknown_fields(self):
|
|
cases = [{'flux': float('nan')}, {'gravity': 'wrong'}, {'oceanzenith': 'unknown'},
|
|
{'physicsfilter': 'gp|exp;command'}, {'invented': 2}, {'vtype': 10}]
|
|
for params in cases:
|
|
with self.subTest(params=params):
|
|
self.assertFalse(self.validate(parameters=params)['ok'])
|
|
self.assertTrue(self.validate(parameters={'physicsfilter': 'gp|exp|sp', 'vegetation': 2})['ok'])
|
|
|
|
def test_nested_objects_and_source_defaults(self):
|
|
result = self.validate(parameters={'glaciers': {'toggle': True}})
|
|
self.assertTrue(result['ok'])
|
|
fields = {f['name']: f for f in self.schema['fields']}
|
|
expected = copy.deepcopy(fields['glaciers']['default'])
|
|
expected['toggle'] = True
|
|
self.assertEqual(result['spec']['parameters']['glaciers'], expected)
|
|
self.assertFalse(self.validate(parameters={'ozone': {'height': 1000}})['ok'])
|
|
self.assertFalse(self.validate(parameters={'glaciers': {'unknown': 2}})['ok'])
|
|
|
|
def test_resource_limits_and_full_year(self):
|
|
self.assertTrue(self.validate(run={'steps': None})['ok'])
|
|
self.assertFalse(self.validate(model={'ncpus': self.config['limits']['max_cpus'] + 1})['ok'])
|
|
self.assertFalse(self.validate(run={'cycles': 0})['ok'])
|
|
self.assertFalse(self.validate(model={'source': '/tmp'})['ok'])
|
|
|
|
def test_surface_recipe_validation_and_php_empty_object(self):
|
|
result = self.validate(parameters=[], surface={'seed': 42})
|
|
self.assertTrue(result['ok'])
|
|
self.assertEqual(result['spec']['parameters'], {})
|
|
self.assertEqual(result['spec']['surface']['seed'], 42)
|
|
self.assertFalse(self.validate(surface={'seed': -1})['ok'])
|
|
self.assertFalse(self.validate(surface={'land_fraction': 2})['ok'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|