# pylint: disable=R0201 from copy import deepcopy, copy from unittest import TestCase from nose.tools import assert_equal, assert_is from mock.mock import Mock from wlauto.exceptions import ConfigError from wlauto.core.configuration.tree import SectionNode from wlauto.core.configuration.configuration import (ConfigurationPoint, Configuration, RunConfiguration, merge_using_priority_specificity, get_type_name) from wlauto.core.configuration.plugin_cache import PluginCache, GENERIC_CONFIGS from wlauto.utils.types import obj_dict # A1 # / \ # B1 B2 # / \ / \ # C1 C2 C3 C4 # \ # D1 a1 = SectionNode({"id": "A1"}) b1 = a1.add_section({"id": "B1"}) b2 = a1.add_section({"id": "B2"}) c1 = b1.add_section({"id": "C1"}) c2 = b1.add_section({"id": "C2"}) c3 = b2.add_section({"id": "C3"}) c4 = b2.add_section({"id": "C4"}) d1 = c2.add_section({"id": "D1"}) DEFAULT_PLUGIN_CONFIG = { "device_config": { "a": { "test3": ["there"], "test5": [5, 4, 3], }, "b": { "test4": 1234, }, }, "some_device": { "a": { "test3": ["how are"], "test2": "MANDATORY", }, "b": { "test3": ["you?"], "test5": [1, 2, 3], } } } def _construct_mock_plugin_cache(values=None): if values is None: values = deepcopy(DEFAULT_PLUGIN_CONFIG) plugin_cache = Mock(spec=PluginCache) plugin_cache.sources = ["a", "b", "c", "d", "e"] def get_plugin_config(plugin_name): return values[plugin_name] plugin_cache.get_plugin_config.side_effect = get_plugin_config def get_plugin_parameters(_): return TestConfiguration.configuration plugin_cache.get_plugin_parameters.side_effect = get_plugin_parameters return plugin_cache class TreeTest(TestCase): def test_node(self): node = SectionNode(1) assert_equal(node.config, 1) assert_is(node.parent, None) assert_equal(node.workload_entries, []) assert_equal(node.children, []) def test_add_workload(self): node = SectionNode(1) node.add_workload(2) assert_equal(len(node.workload_entries), 1) wk = node.workload_entries[0] assert_equal(wk.config, 2) assert_is(wk.parent, node) def test_add_section(self): node = SectionNode(1) new_node = node.add_section(2) assert_equal(len(node.children), 1) assert_is(node.children[0], new_node) assert_is(new_node.parent, node) assert_equal(node.is_leaf, False) assert_equal(new_node.is_leaf, True) def test_descendants(self): for got, expected in zip(b1.descendants(), [c1, d1, c2]): assert_equal(got.config, expected.config) for got, expected in zip(a1.descendants(), [c1, d1, c2, b1, c3, c4, b2]): assert_equal(got.config, expected.config) def test_ancestors(self): for got, expected in zip(d1.ancestors(), [c2, b1, a1]): assert_equal(got.config, expected.config) for _ in a1.ancestors(): raise Exception("A1 is the root, it shouldn't have ancestors") def test_leaves(self): for got, expected in zip(a1.leaves(), [c1, d1, c3, c4]): assert_equal(got.config, expected.config) for got, expected in zip(d1.leaves(), [d1]): assert_equal(got.config, expected.config) def test_source_name(self): assert_equal(a1.name, 'section "A1"') global_section = SectionNode({"id": "global"}) assert_equal(global_section.name, "globally specified configuration") a1.add_workload({'id': 'wk1'}) assert_equal(a1.workload_entries[0].name, 'workload "wk1" from section "A1"') global_section.add_workload({'id': 'wk2'}) assert_equal(global_section.workload_entries[0].name, 'workload "wk2"') class ConfigurationPointTest(TestCase): def test_match(self): cp1 = ConfigurationPoint("test1", aliases=["foo", "bar"]) cp2 = ConfigurationPoint("test2", aliases=["fizz", "buzz"]) assert_equal(cp1.match("test1"), True) assert_equal(cp1.match("foo"), True) assert_equal(cp1.match("bar"), True) assert_equal(cp1.match("fizz"), False) assert_equal(cp1.match("NOT VALID"), False) assert_equal(cp2.match("test2"), True) assert_equal(cp2.match("fizz"), True) assert_equal(cp2.match("buzz"), True) assert_equal(cp2.match("foo"), False) assert_equal(cp2.match("NOT VALID"), False) def test_set_value(self): cp1 = ConfigurationPoint("test", default="hello") cp2 = ConfigurationPoint("test", mandatory=True) cp3 = ConfigurationPoint("test", mandatory=True, default="Hello") cp4 = ConfigurationPoint("test", default=["hello"], merge=True, kind=list) cp5 = ConfigurationPoint("test", kind=int) cp6 = ConfigurationPoint("test5", kind=list, allowed_values=[1, 2, 3, 4, 5]) mock = Mock() mock.name = "ConfigurationPoint Unit Test" # Testing defaults and basic functionality cp1.set_value(mock) assert_equal(mock.test, "hello") cp1.set_value(mock, value="there") assert_equal(mock.test, "there") # Testing mandatory flag err_msg = 'No values specified for mandatory parameter "test" in ' \ 'ConfigurationPoint Unit Test' with self.assertRaisesRegexp(ConfigError, err_msg): cp2.set_value(mock) cp3.set_value(mock) # Should ignore mandatory assert_equal(mock.test, "Hello") # Testing Merging - not in depth that is done in the unit test for merge_config cp4.set_value(mock, value=["there"]) assert_equal(mock.test, ["Hello", "there"]) # Testing type conversion cp5.set_value(mock, value="100") assert_equal(isinstance(mock.test, int), True) msg = 'Bad value "abc" for test; must be an integer' with self.assertRaisesRegexp(ConfigError, msg): cp5.set_value(mock, value="abc") # Testing that validation is not called when no value is set # if it is it will error because it cannot iterate over None cp6.set_value(mock) def test_validation(self): #Test invalid default with self.assertRaises(ValueError): # pylint: disable=W0612 bad_cp = ConfigurationPoint("test", allowed_values=[1], default=100) def is_even(value): if value % 2: return False return True cp1 = ConfigurationPoint("test", kind=int, allowed_values=[1, 2, 3, 4, 5]) cp2 = ConfigurationPoint("test", kind=list, allowed_values=[1, 2, 3, 4, 5]) cp3 = ConfigurationPoint("test", kind=int, constraint=is_even) cp4 = ConfigurationPoint("test", kind=list, mandatory=True, allowed_values=[1, 99]) mock = obj_dict() mock.name = "ConfigurationPoint Validation Unit Test" # Test allowed values cp1.validate_value(mock.name, 1) with self.assertRaises(ConfigError): cp1.validate_value(mock.name, 100) with self.assertRaises(ConfigError): cp1.validate_value(mock.name, [1, 2, 3]) # Test allowed values for lists cp2.validate_value(mock.name, [1, 2, 3]) with self.assertRaises(ConfigError): cp2.validate_value(mock.name, [1, 2, 100]) # Test constraints cp3.validate_value(mock.name, 2) cp3.validate_value(mock.name, 4) cp3.validate_value(mock.name, 6) msg = '"3" failed constraint validation for "test" in "ConfigurationPoint' \ ' Validation Unit Test".' with self.assertRaisesRegexp(ConfigError, msg): cp3.validate_value(mock.name, 3) with self.assertRaises(ValueError): ConfigurationPoint("test", constraint=100) # Test "validate" methods mock.test = None # Mandatory config point not set with self.assertRaises(ConfigError): cp4.validate(mock) cp1.validate(mock) # cp1 doesnt have mandatory set cp4.set_value(mock, value=[99]) cp4.validate(mock) def test_get_type_name(self): def dummy(): pass types = [str, list, int, dummy] names = ["str", "list", "integer", "dummy"] for kind, name in zip(types, names): cp = ConfigurationPoint("test", kind=kind) assert_equal(get_type_name(cp.kind), name) # Subclass to add some config points for use in testing class TestConfiguration(Configuration): name = "Test Config" __configuration = [ ConfigurationPoint("test1", default="hello"), ConfigurationPoint("test2", mandatory=True), ConfigurationPoint("test3", default=["hello"], merge=True, kind=list), ConfigurationPoint("test4", kind=int, default=123), ConfigurationPoint("test5", kind=list, allowed_values=[1, 2, 3, 4, 5]), ] configuration = {cp.name: cp for cp in __configuration} class ConfigurationTest(TestCase): def test_merge_using_priority_specificity(self): # Test good configs plugin_cache = _construct_mock_plugin_cache() expected_result = { "test1": "hello", "test2": "MANDATORY", "test3": ["hello", "there", "how are", "you?"], "test4": 1234, "test5": [1, 2, 3], } result = merge_using_priority_specificity("device_config", "some_device", plugin_cache) assert_equal(result, expected_result) # Test missing mandatory parameter plugin_cache = _construct_mock_plugin_cache(values={ "device_config": { "a": { "test1": "abc", }, }, "some_device": { "b": { "test5": [1, 2, 3], } } }) msg = 'No value specified for mandatory parameter "test2" in some_device.' with self.assertRaisesRegexp(ConfigError, msg): merge_using_priority_specificity("device_config", "some_device", plugin_cache) # Test conflict plugin_cache = _construct_mock_plugin_cache(values={ "device_config": { "e": { 'test2': "NOT_CONFLICTING" } }, "some_device": { 'a': { 'test2': "CONFLICT1" }, 'b': { 'test2': "CONFLICT2" }, 'c': { 'test2': "CONFLICT3" }, }, }) msg = ('Error in "e":\n' '\t"device_config" configuration "test2" has already been specified more specifically for some_device in:\n' '\t\ta\n' '\t\tb\n' '\t\tc') with self.assertRaisesRegexp(ConfigError, msg): merge_using_priority_specificity("device_config", "some_device", plugin_cache) # Test invalid entries plugin_cache = _construct_mock_plugin_cache(values={ "device_config": { "a": { "NOT_A_CFG_POINT": "nope" } }, "some_device": {} }) msg = ('Error in "a":\n\t' 'Invalid entry\(ies\) for "some_device" in "device_config": "NOT_A_CFG_POINT"') with self.assertRaisesRegexp(ConfigError, msg): merge_using_priority_specificity("device_config", "some_device", plugin_cache) plugin_cache = _construct_mock_plugin_cache(values={ "some_device": { "a": { "NOT_A_CFG_POINT": "nope" } }, "device_config": {} }) msg = ('Error in "a":\n\t' 'Invalid entry\(ies\) for "some_device": "NOT_A_CFG_POINT"') with self.assertRaisesRegexp(ConfigError, msg): merge_using_priority_specificity("device_config", "some_device", plugin_cache) # pylint: disable=no-member def test_configuration(self): # Test loading defaults cfg = TestConfiguration() expected = { "test1": "hello", "test3": ["hello"], "test4": 123, } assert_equal(cfg.to_pod(), expected) # If a cfg point is not set an attribute with value None should still be created assert_is(cfg.test2, None) assert_is(cfg.test5, None) # Testing set # Good value cfg.set("test1", "there") assert_equal(cfg.test1, "there") # pylint: disable=E1101 # Unknown value with self.assertRaisesRegexp(ConfigError, 'Unknown Test Config configuration "nope"'): cfg.set("nope", 123) # check_mandatory with self.assertRaises(ConfigError): cfg.set("test2", value=None) cfg.set("test2", value=None, check_mandatory=False) # parameter constraints are tested in the ConfigurationPoint unit test # since this just calls through to `ConfigurationPoint.set_value` # Test validation msg = 'No value specified for mandatory parameter "test2" in Test Config' with self.assertRaisesRegexp(ConfigError, msg): cfg.validate() cfg.set("test2", 1) cfg.validate() # Testing setting values from a dict new_values = { "test1": "This", "test2": "is", "test3": ["a"], "test4": 7357, "test5": [5], } cfg.update_config(new_values) new_values["test3"] = ["hello", "a"] # This cfg point has merge == True for k, v in new_values.iteritems(): assert_equal(getattr(cfg, k), v) #Testing podding pod = cfg.to_pod() new_pod = TestConfiguration.from_pod(copy(pod), None).to_pod() assert_equal(pod, new_pod) #invalid pod entry pod = {'invalid_entry': "nope"} msg = 'Invalid entry\(ies\) for "Test Config": "invalid_entry"' with self.assertRaisesRegexp(ConfigError, msg): TestConfiguration.from_pod(pod, None) #failed pod validation pod = {"test1": "testing"} msg = 'No value specified for mandatory parameter "test2" in Test Config.' with self.assertRaisesRegexp(ConfigError, msg): TestConfiguration.from_pod(pod, None) def test_run_configuration(self): plugin_cache = _construct_mock_plugin_cache() # Test `merge_device_config`` run_config = RunConfiguration() run_config.set("device", "some_device") run_config.merge_device_config(plugin_cache) # Test `to_pod` expected_pod = { "device": "some_device", "device_config": { "test1": "hello", "test2": "MANDATORY", "test3": ["hello", "there", "how are", "you?"], "test4": 1234, "test5": [1, 2, 3], }, "execution_order": "by_iteration", "reboot_policy": "as_needed", "retry_on_status": ['FAILED', 'PARTIAL'], "max_retries": 3, } pod = run_config.to_pod() assert_equal(pod, expected_pod) # Test to_pod > from_pod new_pod = RunConfiguration.from_pod(copy(pod), plugin_cache).to_pod() assert_equal(pod, new_pod) # from_pod with invalid device_config pod['device_config']['invalid_entry'] = "nope" msg = 'Invalid entry "invalid_entry" for device "some_device".' with self.assertRaisesRegexp(ConfigError, msg): RunConfiguration.from_pod(copy(pod), plugin_cache) # from_pod with no device_config pod.pop("device_config") msg = 'No value specified for mandatory parameter "device_config".' with self.assertRaisesRegexp(ConfigError, msg): RunConfiguration.from_pod(copy(pod), plugin_cache) def test_generate_job_spec(self): pass class PluginCacheTest(TestCase): param1 = ConfigurationPoint("param1", aliases="test_global_alias") param2 = ConfigurationPoint("param2", aliases="some_other_alias") param3 = ConfigurationPoint("param3") plugin1 = obj_dict(values={ "name": "plugin 1", "parameters": [ param1, param2, ] }) plugin2 = obj_dict(values={ "name": "plugin 2", "parameters": [ param1, param3, ] }) def get_plugin(self, name): if name == "plugin 1": return self.plugin1 if name == "plugin 2": return self.plugin2 def has_plugin(self, name): return name in ["plugin 1", "plugin 2"] def make_mock_cache(self): mock_loader = Mock() mock_loader.get_plugin_class.side_effect = self.get_plugin mock_loader.list_plugins = Mock(return_value=[self.plugin1, self.plugin2]) mock_loader.has_plugin.side_effect = self.has_plugin return PluginCache(loader=mock_loader) def test_get_params(self): plugin_cache = self.make_mock_cache() expected_params = { self.param1.name: self.param1, self.param2.name: self.param2, } assert_equal(expected_params, plugin_cache.get_plugin_parameters("plugin 1")) def test_global_aliases(self): plugin_cache = self.make_mock_cache() # Check the alias map expected_map = { "plugin 1": { self.param1.aliases: self.param1, self.param2.aliases: self.param2, }, "plugin 2": { self.param1.aliases: self.param1, } } expected_set = set(["test_global_alias", "some_other_alias"]) assert_equal(expected_map, plugin_cache._global_alias_map) assert_equal(expected_set, plugin_cache._list_of_global_aliases) assert_equal(True, plugin_cache.is_global_alias("test_global_alias")) assert_equal(False, plugin_cache.is_global_alias("not_a_global_alias")) # Error when adding to unknown source with self.assertRaises(RuntimeError): plugin_cache.add_global_alias("adding", "too", "early") # Test adding sources for x in xrange(5): plugin_cache.add_source(x) assert_equal([0, 1, 2, 3, 4], plugin_cache.sources) # Error when adding non plugin/global alias/generic with self.assertRaises(RuntimeError): plugin_cache.add_global_alias("unknow_alias", "some_value", 0) # Test adding global alias values plugin_cache.add_global_alias("test_global_alias", "some_value", 0) expected_aliases = {"test_global_alias": {0: "some_value"}} assert_equal(expected_aliases, plugin_cache.global_alias_values) def test_add_config(self): plugin_cache = self.make_mock_cache() # Test adding sources for x in xrange(5): plugin_cache.add_source(x) assert_equal([0, 1, 2, 3, 4], plugin_cache.sources) # Test adding plugin config plugin_cache.add_config("plugin 1", "param1", "some_other_value", 0) expected_plugin_config = {"plugin 1": {0: {"param1": "some_other_value"}}} assert_equal(expected_plugin_config, plugin_cache.plugin_configs) # Test adding generic config for name in GENERIC_CONFIGS: plugin_cache.add_config(name, "param1", "some_value", 0) expected_plugin_config[name] = {} expected_plugin_config[name][0] = {"param1": "some_value"} assert_equal(expected_plugin_config, plugin_cache.plugin_configs) def test_get_plugin_config(self): plugin_cache = self.make_mock_cache() for x in xrange(5): plugin_cache.add_source(x) # Add some global aliases plugin_cache.add_global_alias("test_global_alias", "1", 0) plugin_cache.add_global_alias("test_global_alias", "2", 4) plugin_cache.add_global_alias("test_global_alias", "3", 3) # Test if they are being merged in source order expected_config = { "param1": "2", "param2": None, } assert_equal(expected_config, plugin_cache.get_plugin_config("plugin 1")) # Add some plugin specific config plugin_cache.add_config("plugin 1", "param1", "3", 0) plugin_cache.add_config("plugin 1", "param1", "4", 2) plugin_cache.add_config("plugin 1", "param1", "5", 1) # Test if they are being merged in source order on top of the global aliases expected_config = { "param1": "4", "param2": None, } assert_equal(expected_config, plugin_cache.get_plugin_config("plugin 1")) def test_merge_using_priority_specificity(self): plugin_cache = self.make_mock_cache() for x in xrange(5): plugin_cache.add_source(x) # Add generic configs plugin_cache.add_config("device_config", "param1", '1', 1) plugin_cache.add_config("device_config", "param1", '2', 2) assert_equal(plugin_cache.get_plugin_config("plugin 1", generic_name="device_config"), {"param1": '2', "param2": None}) # Add specific configs at same level as generic config plugin_cache.add_config("plugin 1", "param1", '3', 2) assert_equal(plugin_cache.get_plugin_config("plugin 1", generic_name="device_config"), {"param1": '3', "param2": None}) # Add specific config at higher level plugin_cache.add_config("plugin 1", "param1", '4', 3) assert_equal(plugin_cache.get_plugin_config("plugin 1", generic_name="device_config"), {"param1": '4', "param2": None}) # Add generic config at higher level - should be an error plugin_cache.add_config("device_config", "param1", '5', 4) msg = 'Error in "4":\n' \ '\t"device_config" configuration "param1" has already been specified' \ ' more specifically for plugin 1 in:\n' \ '\t\t2, 3' with self.assertRaisesRegexp(ConfigError, msg): plugin_cache.get_plugin_config("plugin 1", generic_name="device_config")