Merge pull request #52 from salt-formulas/andrewp-overwrite-bug-fix
overwrite bug fix
diff --git a/reclass/core.py b/reclass/core.py
index d3d6187..ed5a392 100644
--- a/reclass/core.py
+++ b/reclass/core.py
@@ -188,7 +188,7 @@
node.interpolate_single_export(q)
except InterpolationError as e:
e.nodename = nodename
- raise InvQueryError(q.contents(), e, context=p, uri=q.uri())
+ raise InvQueryError(q.contents(), e, context=p, uri=q.uri)
inventory[nodename] = node.exports.as_dict()
return inventory
diff --git a/reclass/datatypes/entity.py b/reclass/datatypes/entity.py
index 38360b5..3c927c3 100644
--- a/reclass/datatypes/entity.py
+++ b/reclass/datatypes/entity.py
@@ -83,6 +83,7 @@
self._exports.merge(other._exports)
self._name = other.name
self._uri = other.uri
+ self._parameters._uri = other.uri
if other.environment != None:
self._environment = other.environment
diff --git a/reclass/datatypes/parameters.py b/reclass/datatypes/parameters.py
index e58a1b1..34ccdeb 100644
--- a/reclass/datatypes/parameters.py
+++ b/reclass/datatypes/parameters.py
@@ -27,7 +27,34 @@
from reclass.utils.dictpath import DictPath
from reclass.values.value import Value
from reclass.values.valuelist import ValueList
-from reclass.errors import InfiniteRecursionError, ResolveError, ResolveErrorList, InterpolationError, ParseError, BadReferencesError
+from reclass.errors import InfiniteRecursionError, ResolveError, ResolveErrorList, InterpolationError, BadReferencesError
+
+class ParameterDict(dict):
+ def __init__(self, *args, **kwargs):
+ self._uri = kwargs.pop('uri', None)
+ dict.__init__(self, *args, **kwargs)
+
+ @property
+ def uri(self):
+ return self._uri
+
+ @uri.setter
+ def uri(self, uri):
+ self._uri = uri
+
+
+class ParameterList(list):
+ def __init__(self, *args, **kwargs):
+ self._uri = kwargs.pop('uri', None)
+ list.__init__(self, *args, **kwargs)
+
+ @property
+ def uri(self):
+ return self._uri
+
+ @uri.setter
+ def uri(self, uri):
+ self._uri = uri
class Parameters(object):
@@ -117,10 +144,10 @@
raise
def _wrap_list(self, source, path):
- return [ self._wrap_value(v, path.new_subpath(k)) for (k, v) in enumerate(source) ]
+ return ParameterList([ self._wrap_value(v, path.new_subpath(k)) for (k, v) in enumerate(source) ], uri=self._uri)
def _wrap_dict(self, source, path):
- return { k: self._wrap_value(v, path.new_subpath(k)) for (k, v) in iteritems(source) }
+ return ParameterDict({ k: self._wrap_value(v, path.new_subpath(k)) for (k, v) in iteritems(source) }, uri=self._uri)
def _update_value(self, cur, new):
if isinstance(cur, Value):
@@ -128,14 +155,20 @@
elif isinstance(cur, ValueList):
values = cur
else:
- values = ValueList(Value(cur, self._settings, self._uri), self._settings)
+ uri = self._uri
+ if isinstance(cur, (ParameterDict, ParameterList)):
+ uri = cur.uri
+ values = ValueList(Value(cur, self._settings, uri), self._settings)
if isinstance(new, Value):
values.append(new)
elif isinstance(new, ValueList):
values.extend(new)
else:
- values.append(Value(new, self._settings, self._uri, parse_string=self._parse_strings))
+ uri = self._uri
+ if isinstance(new, (ParameterDict, ParameterList)):
+ uri = new.uri
+ values.append(Value(new, self._settings, uri, parse_string=self._parse_strings))
return values
@@ -167,6 +200,7 @@
ret[key.lstrip(self._settings.dict_key_override_prefix)] = newvalue
else:
ret[key] = self._merge_recurse(ret.get(key), newvalue, path.new_subpath(key))
+
return ret
def _merge_recurse(self, cur, new, path=None):
@@ -213,7 +247,7 @@
if isinstance(other, dict):
wrapped = self._wrap_dict(other, DictPath(self._settings.delimiter))
elif isinstance(other, self.__class__):
- wrapped = self._wrap_dict(other._base, DictPath(self._settings.delimiter))
+ wrapped = other._wrap_dict(other._base, DictPath(self._settings.delimiter))
else:
raise TypeError('Cannot merge %s objects into %s' % (type(other),
self.__class__.__name__))
@@ -224,6 +258,7 @@
if value.is_complex():
p = path.new_subpath(key)
self._unrendered[p] = True
+ container[key] = value
if value.has_inv_query():
self._inv_queries.append((p, value))
if value.needs_all_envs():
@@ -234,29 +269,34 @@
if isinstance(value, Value) and value.is_container():
value = value.contents()
if isinstance(value, dict):
- self._render_simple_dict(value, path.new_subpath(key))
- container[key] = value
+ container[key] = self._render_simple_dict(value, path.new_subpath(key))
elif isinstance(value, list):
- self._render_simple_list(value, path.new_subpath(key))
- container[key] = value
+ container[key] = self._render_simple_list(value, path.new_subpath(key))
elif isinstance(value, Value):
if value.is_complex():
p = path.new_subpath(key)
self._unrendered[p] = True
+ container[key] = value
if value.has_inv_query():
self._inv_queries.append((p, value))
if value.needs_all_envs():
self._needs_all_envs = True
else:
container[key] = value.render(None, None)
+ else:
+ container[key] = value
def _render_simple_dict(self, dictionary, path):
+ new_dict = {}
for (key, value) in iteritems(dictionary):
- self._render_simple_container(dictionary, key, value, path)
+ self._render_simple_container(new_dict, key, value, path)
+ return new_dict
def _render_simple_list(self, item_list, path):
+ new_list = [ None ] * len(item_list)
for n, value in enumerate(item_list):
- self._render_simple_container(item_list, n, value, path)
+ self._render_simple_container(new_list, n, value, path)
+ return new_list
def interpolate(self, inventory=None):
self._initialise_interpolate()
@@ -279,7 +319,7 @@
self._inv_queries = []
self._needs_all_envs = False
self._resolve_errors = ResolveErrorList()
- self._render_simple_dict(self._base, DictPath(self._settings.delimiter))
+ self._base = self._render_simple_dict(self._base, DictPath(self._settings.delimiter))
def _interpolate_inner(self, path, inventory):
value = path.get_value(self._base)
@@ -305,11 +345,14 @@
new = None
else:
raise
+ except InterpolationError as e:
+ e.context = path
+ raise
if isinstance(new, dict):
- self._render_simple_dict(new, path)
+ new = self._render_simple_dict(new, path)
elif isinstance(new, list):
- self._render_simple_list(new, path)
+ new = self._render_simple_list(new, path)
return new
def _interpolate_references(self, path, value, inventory):
@@ -325,7 +368,7 @@
# Therefore, if we encounter False instead of True,
# it means that we have already processed it and are now
# faced with a cyclical reference.
- raise InfiniteRecursionError(path, ref, value.uri())
+ raise InfiniteRecursionError(path, ref, value.uri)
else:
self._interpolate_inner(path_from_ref, inventory)
else:
@@ -345,4 +388,4 @@
old = len(value.get_references())
value.assembleRefs(self._base)
if old == len(value.get_references()):
- raise BadReferencesError(value.get_references(), str(path), value.uri())
+ raise BadReferencesError(value.get_references(), str(path), value.uri)
diff --git a/reclass/datatypes/tests/test_classes.py b/reclass/datatypes/tests/test_classes.py
index 8d396e0..9b9e419 100644
--- a/reclass/datatypes/tests/test_classes.py
+++ b/reclass/datatypes/tests/test_classes.py
@@ -78,8 +78,9 @@
def test_append_invalid_characters(self):
c = Classes()
invalid_name = ' '.join(('foo', 'bar'))
- with self.assertRaises(InvalidClassnameError):
+ with self.assertRaises(InvalidClassnameError) as e:
c.append_if_new(invalid_name)
+ self.assertEqual(e.exception.message, "Invalid character ' ' in class name 'foo bar'.")
def test_merge_unique(self):
c = Classes(TESTLIST1)
diff --git a/reclass/datatypes/tests/test_entity.py b/reclass/datatypes/tests/test_entity.py
index b09e904..c08a500 100644
--- a/reclass/datatypes/tests/test_entity.py
+++ b/reclass/datatypes/tests/test_entity.py
@@ -253,10 +253,11 @@
node1_entity.initialise_interpolation()
node2_entity.initialise_interpolation()
queries = node1_entity.parameters.get_inv_queries()
- with self.assertRaises(ResolveError):
+ with self.assertRaises(ResolveError) as e:
for p, q in queries:
node1_entity.interpolate_single_export(q)
node2_entity.interpolate_single_export(q)
+ self.assertEqual(e.exception.message, "-> \n Cannot resolve ${b}, at a")
def test_exports_failed_render_ignore(self):
node1_exports = Exports({'a': '${a}'}, SETTINGS, '')
diff --git a/reclass/datatypes/tests/test_exports.py b/reclass/datatypes/tests/test_exports.py
index 6a6dcde..caa0522 100644
--- a/reclass/datatypes/tests/test_exports.py
+++ b/reclass/datatypes/tests/test_exports.py
@@ -62,9 +62,10 @@
def test_list_if_expr_invquery(self):
e = {'node1': {'a': 1, 'b': 2}, 'node2': {'a': 3, 'b': 3}, 'node3': {'a': 3, 'b': 2}}
p = Parameters({'exp': '$[ if exports:b == 2 ]'}, SETTINGS, '')
- r = {'exp': ['node1', 'node3']}
+ r1 = {'exp': ['node1', 'node3']}
+ r2 = {'exp': ['node3', 'node1']}
p.interpolate(e)
- self.assertEqual(p.as_dict(), r)
+ self.assertIn(p.as_dict(), [ r1, r2 ])
def test_if_expr_invquery_wth_and(self):
e = {'node1': {'a': 1, 'b': 4, 'c': False}, 'node2': {'a': 3, 'b': 4, 'c': True}}
@@ -97,9 +98,10 @@
def test_list_if_expr_invquery_with_and(self):
e = {'node1': {'a': 1, 'b': 2}, 'node2': {'a': 3, 'b': 3}, 'node3': {'a': 3, 'b': 4}}
p = Parameters({'exp': '$[ if exports:b == 2 or exports:b == 4 ]'}, SETTINGS, '')
- r = {'exp': ['node1', 'node3']}
+ r1 = {'exp': ['node1', 'node3']}
+ r2 = {'exp': ['node3', 'node1']}
p.interpolate(e)
- self.assertEqual(p.as_dict(), r)
+ self.assertIn(p.as_dict(), [ r1, r2 ])
if __name__ == '__main__':
unittest.main()
diff --git a/reclass/datatypes/tests/test_parameters.py b/reclass/datatypes/tests/test_parameters.py
index fb4a11b..b15f8ce 100644
--- a/reclass/datatypes/tests/test_parameters.py
+++ b/reclass/datatypes/tests/test_parameters.py
@@ -17,7 +17,9 @@
from reclass.settings import Settings
from reclass.datatypes import Parameters
-from reclass.errors import InfiniteRecursionError, InterpolationError, ResolveError, ResolveErrorList
+from reclass.values.value import Value
+from reclass.values.scaitem import ScaItem
+from reclass.errors import InfiniteRecursionError, InterpolationError, ResolveError, ResolveErrorList, TypeMergeError
import unittest
try:
@@ -120,13 +122,17 @@
self.assertEqual(b1.__eq__.call_count, 0)
def test_construct_wrong_type(self):
- with self.assertRaises(TypeError):
- self._construct_mocked_params('wrong type')
+ with self.assertRaises(TypeError) as e:
+ self._construct_mocked_params(str('wrong type'))
+ self.assertIn(str(e.exception), [ "Cannot merge <type 'str'> objects into Parameters", # python 2
+ "Cannot merge <class 'str'> objects into Parameters" ]) # python 3
def test_merge_wrong_type(self):
p, b = self._construct_mocked_params()
- with self.assertRaises(TypeError):
- p.merge('wrong type')
+ with self.assertRaises(TypeError) as e:
+ p.merge(str('wrong type'))
+ self.assertIn(str(e.exception), [ "Cannot merge <type 'str'> objects into Parameters", # python 2
+ "Cannot merge <class 'str'> objects into Parameters"]) # python 3
def test_get_dict(self):
p, b = self._construct_mocked_params(SIMPLE)
@@ -138,12 +144,8 @@
mergee = {'five':5,'four':4,'None':None,'tuple':(1,2,3)}
p2, b2 = self._construct_mocked_params(mergee)
p1.merge(p2)
- p1.initialise_interpolation()
- for (key, value) in iteritems(mergee):
- # check that each key, value in mergee resulted in a get call and
- # a __setitem__ call against b1 (the merge target)
- self.assertIn(mock.call(key), b1.get.call_args_list)
- self.assertIn(mock.call(key, value), b1.__setitem__.call_args_list)
+ self.assertEqual(b1.get.call_count, 4)
+ self.assertEqual(b1.__setitem__.call_count, 4)
def test_stray_occurrence_overwrites_during_interpolation(self):
p1 = Parameters({'r' : mock.sentinel.ref, 'b': '${r}'}, SETTINGS, '')
@@ -183,57 +185,112 @@
self.assertListEqual(p1.as_dict()['list'], l1+l2)
def test_merge_list_into_scalar(self):
+ l = ['foo', 1, 2]
+ p1 = Parameters(dict(key=l[0]), SETTINGS, '')
+ p2 = Parameters(dict(key=l[1:]), SETTINGS, '')
+ with self.assertRaises(TypeMergeError) as e:
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge list over scalar, at key, in ; ")
+
+ def test_merge_list_into_scalar_allow(self):
settings = Settings({'allow_list_over_scalar': True})
l = ['foo', 1, 2]
p1 = Parameters(dict(key=l[0]), settings, '')
p2 = Parameters(dict(key=l[1:]), settings, '')
p1.merge(p2)
- p1.initialise_interpolation()
+ p1.interpolate()
self.assertListEqual(p1.as_dict()['key'], l)
def test_merge_scalar_over_list(self):
l = ['foo', 1, 2]
+ p1 = Parameters(dict(key=l[:2]), SETTINGS, '')
+ p2 = Parameters(dict(key=l[2]), SETTINGS, '')
+ with self.assertRaises(TypeMergeError) as e:
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge scalar over list, at key, in ; ")
+
+ def test_merge_scalar_over_list_allow(self):
+ l = ['foo', 1, 2]
settings = Settings({'allow_scalar_over_list': True})
p1 = Parameters(dict(key=l[:2]), settings, '')
p2 = Parameters(dict(key=l[2]), settings, '')
p1.merge(p2)
- p1.initialise_interpolation()
+ p1.interpolate()
self.assertEqual(p1.as_dict()['key'], l[2])
def test_merge_none_over_list(self):
l = ['foo', 1, 2]
+ settings = Settings({'allow_none_override': False})
+ p1 = Parameters(dict(key=l[:2]), settings, '')
+ p2 = Parameters(dict(key=None), settings, '')
+ with self.assertRaises(TypeMergeError) as e:
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge scalar over list, at key, in ; ")
+
+ def test_merge_none_over_list_allow(self):
+ l = ['foo', 1, 2]
settings = Settings({'allow_none_override': True})
p1 = Parameters(dict(key=l[:2]), settings, '')
p2 = Parameters(dict(key=None), settings, '')
p1.merge(p2)
- p1.initialise_interpolation()
+ p1.interpolate()
self.assertEqual(p1.as_dict()['key'], None)
- def test_merge_none_over_list_negative(self):
- l = ['foo', 1, 2]
- settings = Settings({'allow_none_override': False})
- p1 = Parameters(dict(key=l[:2]), settings, '')
- p2 = Parameters(dict(key=None), settings, '')
- with self.assertRaises(TypeError):
+ def test_merge_dict_over_scalar(self):
+ d = { 'one': 1, 'two': 2 }
+ p1 = Parameters({ 'a': 1 }, SETTINGS, '')
+ p2 = Parameters({ 'a': d }, SETTINGS, '')
+ with self.assertRaises(TypeMergeError) as e:
p1.merge(p2)
- p1.initialise_interpolation()
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge dictionary over scalar, at a, in ; ")
+
+ def test_merge_dict_over_scalar_allow(self):
+ settings = Settings({'allow_dict_over_scalar': True})
+ d = { 'one': 1, 'two': 2 }
+ p1 = Parameters({ 'a': 1 }, settings, '')
+ p2 = Parameters({ 'a': d }, settings, '')
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(p1.as_dict(), { 'a': d })
+
+ def test_merge_scalar_over_dict(self):
+ d = { 'one': 1, 'two': 2}
+ p1 = Parameters({ 'a': d }, SETTINGS, '')
+ p2 = Parameters({ 'a': 1 }, SETTINGS, '')
+ with self.assertRaises(TypeMergeError) as e:
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge scalar over dictionary, at a, in ; ")
+
+ def test_merge_scalar_over_dict_allow(self):
+ d = { 'one': 1, 'two': 2}
+ settings = Settings({'allow_scalar_over_dict': True})
+ p1 = Parameters({ 'a': d }, settings, '')
+ p2 = Parameters({ 'a': 1 }, settings, '')
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(p1.as_dict(), { 'a': 1})
def test_merge_none_over_dict(self):
+ p1 = Parameters(dict(key=SIMPLE), SETTINGS, '')
+ p2 = Parameters(dict(key=None), SETTINGS, '')
+ with self.assertRaises(TypeMergeError) as e:
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(e.exception.message, "-> \n Canot merge scalar over dictionary, at key, in ; ")
+
+ def test_merge_none_over_dict_allow(self):
settings = Settings({'allow_none_override': True})
p1 = Parameters(dict(key=SIMPLE), settings, '')
p2 = Parameters(dict(key=None), settings, '')
p1.merge(p2)
- p1.initialise_interpolation()
+ p1.interpolate()
self.assertEqual(p1.as_dict()['key'], None)
- def test_merge_none_over_dict_negative(self):
- settings = Settings({'allow_none_override': False})
- p1 = Parameters(dict(key=SIMPLE), settings, '')
- p2 = Parameters(dict(key=None), settings, '')
- with self.assertRaises(TypeError):
- p1.merge(p2)
- p1.initialise_interpolation()
-
# def test_merge_bare_dict_over_dict(self):
# settings = Settings({'allow_bare_override': True})
# p1 = Parameters(dict(key=SIMPLE), settings, '')
@@ -285,22 +342,6 @@
p.initialise_interpolation()
self.assertDictEqual(p.as_dict(), dict(dict=goal))
- def test_merge_dict_into_scalar(self):
- p = Parameters(dict(base='foo'), SETTINGS, '')
- p2 = Parameters(dict(base=SIMPLE), SETTINGS, '')
- with self.assertRaises(TypeError):
- p.merge(p2)
- p.interpolate()
-
- def test_merge_scalar_over_dict(self):
- settings = Settings({'allow_scalar_over_dict': True})
- p = Parameters(dict(base=SIMPLE), settings, '')
- mergee = {'base':'foo'}
- p2 = Parameters(mergee, settings, '')
- p.merge(p2)
- p.initialise_interpolation()
- self.assertDictEqual(p.as_dict(), mergee)
-
def test_interpolate_single(self):
v = 42
d = {'foo': 'bar'.join(SETTINGS.reference_sentinels),
@@ -340,8 +381,11 @@
d = {'foo': 'bar'.join(SETTINGS.reference_sentinels),
'bar': 'foo'.join(SETTINGS.reference_sentinels)}
p = Parameters(d, SETTINGS, '')
- with self.assertRaises(InfiniteRecursionError):
+ with self.assertRaises(InfiniteRecursionError) as e:
p.interpolate()
+ # interpolation can start with foo or bar
+ self.assertIn(e.exception.message, [ "-> \n Infinite recursion: ${foo}, at bar",
+ "-> \n Infinite recursion: ${bar}, at foo"])
def test_nested_references(self):
d = {'a': '${${z}}', 'b': 2, 'z': 'b'}
@@ -423,6 +467,22 @@
p1.interpolate()
self.assertEqual(p1.as_dict(), r)
+ def test_overwrite_dict(self):
+ p1 = Parameters({'a': { 'one': 1, 'two': 2 }}, SETTINGS, '')
+ p2 = Parameters({'~a': { 'three': 3, 'four': 4 }}, SETTINGS, '')
+ r = {'a': { 'three': 3, 'four': 4 }}
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(p1.as_dict(), r)
+
+ def test_overwrite_list(self):
+ p1 = Parameters({'a': [1, 2]}, SETTINGS, '')
+ p2 = Parameters({'~a': [3, 4]}, SETTINGS, '')
+ r = {'a': [3, 4]}
+ p1.merge(p2)
+ p1.interpolate()
+ self.assertEqual(p1.as_dict(), r)
+
def test_interpolate_escaping(self):
v = 'bar'.join(SETTINGS.reference_sentinels)
d = {'foo': SETTINGS.escape_character + 'bar'.join(SETTINGS.reference_sentinels),
@@ -575,7 +635,9 @@
p1 = Parameters({'alpha': '${gamma}', 'beta': '${gamma}'}, SETTINGS, '')
with self.assertRaises(ResolveErrorList) as error:
p1.interpolate()
- self.assertEqual(error.exception.message, "-> \n Cannot resolve ${gamma}, at alpha\n Cannot resolve ${gamma}, at beta")
+ # interpolation can start with either alpha or beta
+ self.assertIn(error.exception.message, [ "-> \n Cannot resolve ${gamma}, at alpha\n Cannot resolve ${gamma}, at beta",
+ "-> \n Cannot resolve ${gamma}, at beta\n Cannot resolve ${gamma}, at alpha"])
def test_force_single_resolve_error(self):
settings = copy.deepcopy(SETTINGS)
@@ -583,7 +645,9 @@
p1 = Parameters({'alpha': '${gamma}', 'beta': '${gamma}'}, settings, '')
with self.assertRaises(ResolveError) as error:
p1.interpolate()
- self.assertEqual(error.exception.message, "-> \n Cannot resolve ${gamma}, at alpha")
+ # interpolation can start with either alpha or beta
+ self.assertIn(error.exception.message, [ "-> \n Cannot resolve ${gamma}, at alpha",
+ "-> \n Cannot resolve ${gamma}, at beta"])
def test_ignore_overwriten_missing_reference(self):
settings = copy.deepcopy(SETTINGS)
diff --git a/reclass/errors.py b/reclass/errors.py
index 800a2f8..fdd7f38 100644
--- a/reclass/errors.py
+++ b/reclass/errors.py
@@ -208,6 +208,7 @@
msg = 'Cannot resolve {0}'.format(self.reference.join(REFERENCE_SENTINELS)) + self._add_context_and_uri()
return [ msg ]
+
class ResolveErrorList(InterpolationError):
def __init__(self):
super(ResolveErrorList, self).__init__(msg=None)
@@ -280,6 +281,18 @@
return msg
+class TypeMergeError(InterpolationError):
+
+ def __init__(self, value1, value2, uri):
+ super(TypeMergeError, self).__init__(msg=None, uri=uri, tbFlag=False)
+ self.type1 = value1.item_type_str()
+ self.type2 = value2.item_type_str()
+
+ def _get_error_message(self):
+ msg = [ 'Canot merge {0} over {1}'.format(self.type1, self.type2) + self._add_context_and_uri() ]
+ return msg
+
+
class ExpressionError(InterpolationError):
def __init__(self, msg, rc=posix.EX_DATAERR, tbFlag=True):
diff --git a/reclass/utils/dictpath.py b/reclass/utils/dictpath.py
index 39b9572..57971b4 100644
--- a/reclass/utils/dictpath.py
+++ b/reclass/utils/dictpath.py
@@ -61,17 +61,17 @@
def __init__(self, delim, contents=None):
self._delim = delim
+
if contents is None:
self._parts = []
+ elif isinstance(contents, list):
+ self._parts = contents
+ elif isinstance(contents, six.string_types):
+ self._parts = self._split_string(contents)
+ elif isinstance(contents, tuple):
+ self._parts = list(contents)
else:
- if isinstance(contents, list):
- self._parts = contents
- elif isinstance(contents, six.string_types):
- self._parts = self._split_string(contents)
- elif isinstance(contents, tuple):
- self._parts = list(contents)
- else:
- raise TypeError('DictPath() takes string or list, '\
+ raise TypeError('DictPath() takes string or list, '\
'not %s' % type(contents))
def __repr__(self):
diff --git a/reclass/values/compitem.py b/reclass/values/compitem.py
index 7928e6f..183bc43 100644
--- a/reclass/values/compitem.py
+++ b/reclass/values/compitem.py
@@ -46,17 +46,7 @@
def merge_over(self, item):
if item.type == Item.SCALAR or item.type == Item.COMPOSITE:
return self
- elif item.type == Item.LIST:
- if self._settings.allow_scalar_over_list or (self._settings.allow_none_override and self._value in [None, 'none', 'None']):
- return self
- else:
- raise TypeError('allow scalar over list = False: cannot merge %s over %s' % (repr(self), repr(item)))
- elif item.type == Item.DICTIONARY:
- if self._settings.allow_scalar_over_dict or (self._settings.allow_none_override and self._value in [None, 'none', 'None']):
- return self
- else:
- raise TypeError('allow scalar over dict = False: cannot merge %s over %s' % (repr(self), repr(item)))
- raise TypeError('Cannot merge %s over %s' % (repr(self), repr(item)))
+ raise RuntimeError('Trying to merge %s over %s' % (repr(self), repr(item)))
def render(self, context, inventory):
# Preserve type if only one item
diff --git a/reclass/values/dictitem.py b/reclass/values/dictitem.py
index 76cefe2..d5272b9 100644
--- a/reclass/values/dictitem.py
+++ b/reclass/values/dictitem.py
@@ -24,14 +24,6 @@
def is_container(self):
return True
- def merge_over(self, item):
- if item.type == Item.SCALAR:
- if item.contents() is None or self._settings.allow_dict_over_scalar:
- return self
- else:
- raise TypeError('allow dict over scalar = False: cannot merge %s onto %s' % (repr(self), repr(item)))
- raise TypeError('Cannot merge %s over %s' % (repr(self), repr(item)))
-
def render(self, context, inventory):
return self._dict
diff --git a/reclass/values/item.py b/reclass/values/item.py
index bc507f4..cad3684 100644
--- a/reclass/values/item.py
+++ b/reclass/values/item.py
@@ -19,6 +19,10 @@
REFERENCE = 5
SCALAR = 6
+ TYPE_STR = { COMPOSITE: 'composite', DICTIONARY: 'dictionary',
+ INV_QUERY: 'invventory query', LIST: 'list',
+ REFERENCE: 'reference', SCALAR: 'scalar' }
+
def allRefs(self):
return True
@@ -45,3 +49,6 @@
def render(self, context, exports):
msg = "Item class {0} does not implement render()"
raise NotImplementedError(msg.format(self.__class__.__name__))
+
+ def type_str(self):
+ return self.TYPE_STR[self.type]
diff --git a/reclass/values/listitem.py b/reclass/values/listitem.py
index 8f1a21d..41c02dd 100644
--- a/reclass/values/listitem.py
+++ b/reclass/values/listitem.py
@@ -31,15 +31,7 @@
if item.type == Item.LIST:
item._list.extend(self._list)
return item
- elif item.type == Item.SCALAR:
- if item.contents() is None:
- return self
- elif self._settings.allow_list_over_scalar:
- self._list.insert(0, item.contents())
- return self
- else:
- raise TypeError('allow list over scalar = False: cannot merge %s onto %s' % (repr(self), repr(item)))
- raise TypeError('Cannot merge %s over %s' % (repr(self), repr(item)))
+ raise RuntimeError('Trying to merge %s over %s' % (repr(self), repr(item)))
def __repr__(self):
return 'ListItem(%r)' % (self._list)
diff --git a/reclass/values/scaitem.py b/reclass/values/scaitem.py
index f3057cb..c16ab45 100644
--- a/reclass/values/scaitem.py
+++ b/reclass/values/scaitem.py
@@ -24,17 +24,7 @@
def merge_over(self, item):
if item.type == Item.SCALAR or item.type == Item.COMPOSITE:
return self
- elif item.type == Item.LIST:
- if self._settings.allow_scalar_over_list or (self._settings.allow_none_override and self._value is None):
- return self
- else:
- raise TypeError('allow scalar over list = False: cannot merge %s over %s' % (repr(self), repr(item)))
- elif item.type == Item.DICTIONARY:
- if self._settings.allow_scalar_over_dict or (self._settings.allow_none_override and self._value is None):
- return self
- else:
- raise TypeError('allow scalar over dict = False: cannot merge %s over %s' % (repr(self), repr(item)))
- raise TypeError('Cannot merge %s over %s' % (repr(self), repr(item)))
+ raise RuntimeError('Trying to merge %s over %s' % (repr(self), repr(item)))
def render(self, context, inventory):
return self._value
diff --git a/reclass/values/value.py b/reclass/values/value.py
index f87a9b4..286407c 100644
--- a/reclass/values/value.py
+++ b/reclass/values/value.py
@@ -48,9 +48,16 @@
def overwrite(self, overwrite):
self._overwrite = overwrite
+ @property
def uri(self):
return self._uri
+ def item_type(self):
+ return self._item.type
+
+ def item_type_str(self):
+ return self._item.type_str()
+
def is_container(self):
return self._item.is_container()
diff --git a/reclass/values/valuelist.py b/reclass/values/valuelist.py
index 8b8dd51..adffb87 100644
--- a/reclass/values/valuelist.py
+++ b/reclass/values/valuelist.py
@@ -13,7 +13,7 @@
import copy
import sys
-from reclass.errors import ResolveError
+from reclass.errors import ResolveError, TypeMergeError
class ValueList(object):
@@ -25,8 +25,13 @@
self._inv_refs = []
self._has_inv_query = False
self._ignore_failed_render = False
+ self._is_complex = False
self._update()
+ @property
+ def uri(self):
+ return '; '.join([ str(x.uri) for x in self._values ])
+
def append(self, value):
self._values.append(value)
self._update()
@@ -38,9 +43,11 @@
def _update(self):
self.assembleRefs()
self._check_for_inv_query()
-
- def uri(self):
- return '; '.join([ x.uri() for x in self._values ])
+ self._is_complex = False
+ item_type = self._values[0].item_type()
+ for v in self._values:
+ if v.is_complex() or v.overwrite or v.item_type() != item_type:
+ self._is_complex = True
def has_references(self):
return len(self._refs) > 0
@@ -52,7 +59,7 @@
return self._inv_refs
def is_complex(self):
- return (self.has_references() | self.has_inv_query())
+ return self._is_complex
def get_references(self):
return self._refs
@@ -104,6 +111,8 @@
try:
new = value.render(context, inventory)
except ResolveError as e:
+ # only ignore failed renders if ignore_overwritten_missing_references is set and we are dealing with a scalar value
+ # and it's not the last item in the values list
if self._settings.ignore_overwritten_missing_references and not isinstance(output, (dict, list)) and n != (len(self._values)-1):
new = None
last_error = e
@@ -115,23 +124,51 @@
output = new
deepCopied = False
else:
- if isinstance(output, dict) and isinstance(new, dict):
- p1 = Parameters(output, self._settings, None, parse_strings=False)
- p2 = Parameters(new, self._settings, None, parse_strings=False)
- p1.merge(p2)
- output = p1.as_dict()
- continue
- elif isinstance(output, list) and isinstance(new, list):
- if not deepCopied:
- output = copy.deepcopy(output)
- deepCopied = True
- output.extend(new)
- continue
- elif isinstance(output, (dict, list)) or isinstance(new, (dict, list)):
- raise TypeError('Cannot merge %s over %s' % (repr(self._values[n]), repr(self._values[n-1])))
+ if isinstance(output, dict):
+ if isinstance(new, dict):
+ p1 = Parameters(output, self._settings, None, parse_strings=False)
+ p2 = Parameters(new, self._settings, None, parse_strings=False)
+ p1.merge(p2)
+ output = p1.as_dict()
+ elif isinstance(new, list):
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
+ elif self._settings.allow_scalar_over_dict or (self._settings.allow_none_override and new is None):
+ output = new
+ deepCopied = False
+ else:
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
+ elif isinstance(output, list):
+ if isinstance(new, list):
+ if not deepCopied:
+ output = copy.deepcopy(output)
+ deepCopied = True
+ output.extend(new)
+ elif isinstance(new, dict):
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
+ elif self._settings.allow_scalar_over_list or (self._settings.allow_none_override and new is None):
+ output = new
+ deepCopied = False
+ else:
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
else:
- output = new
- deepCopied = False
+ if isinstance(new, dict):
+ if self._settings.allow_dict_over_scalar:
+ output = new
+ deepCopied = False
+ else:
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
+ elif isinstance(new, list):
+ if self._settings.allow_list_over_scalar:
+ output_list = list()
+ output_list.append(output)
+ output_list.extend(new)
+ output = output_list
+ deepCopied = True
+ else:
+ raise TypeMergeError(self._values[n], self._values[n-1], self.uri)
+ else:
+ output = new
+ deepCopied = False
if isinstance(output, (dict, list)) and last_error is not None:
raise last_error