180 lines
6.6 KiB
Python
180 lines
6.6 KiB
Python
"""
|
|
@package tests.test_project
|
|
unit tests for pmsco.project.
|
|
|
|
the purpose of these tests is to help debugging the code.
|
|
|
|
to run the tests, change to the directory which contains the tests directory, and execute =nosetests=.
|
|
|
|
@pre nose and mock must be installed.
|
|
|
|
@author Matthias Muntwiler, matthias.muntwiler@psi.ch
|
|
|
|
@copyright (c) 2015-25 by Paul Scherrer Institut @n
|
|
Licensed under the Apache License, Version 2.0 (the "License"); @n
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
"""
|
|
|
|
import mock
|
|
import numpy as np
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
import pmsco.data as data
|
|
import pmsco.dispatch as dispatch
|
|
import pmsco.project as project
|
|
from pmsco.scan import Scan
|
|
|
|
|
|
class TestModelSpace(unittest.TestCase):
|
|
def setUp(self):
|
|
self.d1 = {
|
|
"A": {"start": 2.1, "min": 2.0, "max": 3.0, "step": 0.05},
|
|
"B": {"start": 15.0, "min": 0.0, "max": 30.0, "step": 1.0}}
|
|
self.d2 = {
|
|
"C": {"start": 22.0, "min": 15.0, "max": 25.0, "step": 1.0},
|
|
"D": {"start": 1.5, "min": 0.5, "max": 2.0, "step": 0.25}}
|
|
|
|
def test_eval_param_value(self):
|
|
dummy_value = 15.3
|
|
ms = project.ModelSpace()
|
|
ms.project_symbols = {'numpy': np}
|
|
self.assertAlmostEqual(ms._eval_param_value(0.01), 0.01)
|
|
self.assertAlmostEqual(ms._eval_param_value('0.01'), 0.01)
|
|
self.assertAlmostEqual(ms._eval_param_value('numpy.sin(0.1)'), np.sin(0.1))
|
|
self.assertAlmostEqual(ms._eval_param_value('abs(-0.1)'), 0.1)
|
|
self.assertTrue(np.isnan(ms._eval_param_value(None)))
|
|
self.assertTrue(np.isnan(ms._eval_param_value(np.nan)))
|
|
self.assertRaises(ValueError, ms._eval_param_value, '')
|
|
# should not have access to local symbols
|
|
self.assertRaises(NameError, ms._eval_param_value, 'dummy_value')
|
|
|
|
def test_add_param(self):
|
|
ms = project.ModelSpace()
|
|
ms.start['A'] = 2.1
|
|
ms.min['A'] = 2.0
|
|
ms.max['A'] = 3.0
|
|
ms.step['A'] = 0.05
|
|
ms.add_param("E", 5.0, 1.0, 9.0, 0.2)
|
|
ms.add_param("F", 8.0, width=6.0, step=0.5)
|
|
d_start = {'A': 2.1, 'E': 5.0, 'F': 8.0}
|
|
d_min = {'A': 2.0, 'E': 1.0, 'F': 5.0}
|
|
d_max = {'A': 3.0, 'E': 9.0, 'F': 11.0}
|
|
d_step = {'A': 0.05, 'E': 0.2, 'F': 0.5}
|
|
self.assertDictEqual(ms.start, d_start)
|
|
self.assertDictEqual(ms.min, d_min)
|
|
self.assertDictEqual(ms.max, d_max)
|
|
self.assertDictEqual(ms.step, d_step)
|
|
|
|
def test_get_param(self):
|
|
ms = project.ModelSpace()
|
|
ms.add_param("A", **self.d1['A'])
|
|
ms.add_param("B", **self.d1['B'])
|
|
result = ms.get_param('B')
|
|
expected = {'start': 15.0, 'min': 0.0, 'max': 30.0, 'step': 1.0}
|
|
self.assertIsInstance(result, project.ParamSpace)
|
|
self.assertEqual(result.start, expected['start'])
|
|
self.assertEqual(result.min, expected['min'])
|
|
self.assertEqual(result.max, expected['max'])
|
|
self.assertEqual(result.step, expected['step'])
|
|
|
|
def test_set_param_dict(self):
|
|
ms = project.ModelSpace()
|
|
ms.set_param_dict(self.d1)
|
|
ms.set_param_dict(self.d2)
|
|
d_start = {'C': 22.0, 'D': 1.5}
|
|
d_min = {'C': 15.0, 'D': 0.5}
|
|
d_max = {'C': 25.0, 'D': 2.0}
|
|
d_step = {'C': 1.0, 'D': 0.25}
|
|
self.assertDictEqual(ms.start, d_start)
|
|
self.assertDictEqual(ms.min, d_min)
|
|
self.assertDictEqual(ms.max, d_max)
|
|
self.assertDictEqual(ms.step, d_step)
|
|
|
|
|
|
class TestProject(unittest.TestCase):
|
|
def setUp(self):
|
|
# before each test method
|
|
self.project = project.Project()
|
|
|
|
def tearDown(self):
|
|
# after each test method
|
|
pass
|
|
|
|
@classmethod
|
|
def setup_class(cls):
|
|
# before any methods in this class
|
|
pass
|
|
|
|
@classmethod
|
|
def teardown_class(cls):
|
|
# teardown_class() after any methods in this class
|
|
pass
|
|
|
|
def test_resolve_directories(self):
|
|
self.project.job_name = "jn1"
|
|
self.project.job_tags['jt1'] = 'tag1'
|
|
self.project.directories['report'] = Path("${output}/reports")
|
|
self.project.directories['output'] = Path("${home}/test/output")
|
|
self.project.directories.resolve_directories(check=True)
|
|
|
|
expected = Path(Path.home(), "test", "output")
|
|
self.assertEqual(expected, self.project.directories['output'])
|
|
expected = Path(Path.home(), "test", "output", "reports")
|
|
self.assertEqual(expected, self.project.directories['report'])
|
|
|
|
def test_resolve_path(self):
|
|
self.project.job_name = "jn1"
|
|
self.project.job_tags['jt1'] = 'tag1'
|
|
self.project.directories['output'] = Path.home() / "test" / "output"
|
|
self.project.directories['report'] = self.project.directories['output'] / "reports"
|
|
extra = {'param_name': 'A', 'value': 25.6}
|
|
template = "${report}/${job_name}-${jt1}-${param_name}"
|
|
rps = self.project.directories.resolve_path(template, extra)
|
|
rpp = self.project.directories.resolve_path(Path(template), extra)
|
|
expected = Path(Path.home(), "test", "output", "reports", "jn1-tag1-A")
|
|
self.assertEqual(str(expected), rps)
|
|
self.assertEqual(expected, rpp)
|
|
|
|
fdict = {'base': rpp.stem, 'gen': 14, 'param0': 'A', 'param1': 'B'}
|
|
template = "my_calc gen ${gen}"
|
|
title = self.project.directories.resolve_path(template, fdict)
|
|
expected = "my_calc gen 14"
|
|
self.assertEqual(expected, title)
|
|
|
|
@mock.patch('pmsco.data.load_data')
|
|
@mock.patch('pmsco.data.save_data')
|
|
def test_combine_domains(self, save_data_mock, load_data_mock):
|
|
self.project.scans.append(Scan())
|
|
|
|
parent_task = dispatch.CalculationTask()
|
|
parent_task.change_id(model=0, scan=0)
|
|
parent_task.model['wdom1'] = 0.5
|
|
|
|
child_tasks = [parent_task.copy()] * 2
|
|
for idx, task in enumerate(child_tasks):
|
|
task.change_id(domain=idx)
|
|
|
|
data1 = data.create_data(5, datatype='EI')
|
|
data1['e'] = np.arange(5)
|
|
data1['i'] = 10.
|
|
data2 = data1.copy()
|
|
data2['i'] = 10.
|
|
load_data_mock.side_effect = [data1, data2]
|
|
data3 = data1.copy()
|
|
data3['i'] = (10. + 0.5 * 10.) / 1.5
|
|
|
|
self.project.combine_domains(parent_task, child_tasks)
|
|
|
|
save_data_mock.assert_called()
|
|
args, kwargs = save_data_mock.call_args
|
|
result_data = args[1]
|
|
np.testing.assert_array_almost_equal(result_data['e'], data3['e'])
|
|
np.testing.assert_array_almost_equal(result_data['i'], data3['i'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|