mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
Migrate unit.serializers
to PyTest
This commit is contained in:
parent
0156fb16e7
commit
84fdfca253
6 changed files with 399 additions and 386 deletions
|
@ -214,7 +214,7 @@ salt/runners/venafiapi.py:
|
|||
- integration.externalapi.test_venafiapi
|
||||
|
||||
salt/serializers/*:
|
||||
- unit.serializers.test_serializers
|
||||
- pytests.unit.serializers.test_serializers
|
||||
|
||||
salt/(cli/spm\.py|spm/.+):
|
||||
- unit.test_spm
|
||||
|
|
0
tests/pytests/unit/serializers/__init__.py
Normal file
0
tests/pytests/unit/serializers/__init__.py
Normal file
397
tests/pytests/unit/serializers/test_serializers.py
Normal file
397
tests/pytests/unit/serializers/test_serializers.py
Normal file
|
@ -0,0 +1,397 @@
|
|||
from textwrap import dedent
|
||||
|
||||
import jinja2
|
||||
import pytest
|
||||
import salt.serializers.configparser as configparser
|
||||
import salt.serializers.json as json
|
||||
import salt.serializers.msgpack as msgpack
|
||||
import salt.serializers.plist as plist
|
||||
import salt.serializers.python as python
|
||||
import salt.serializers.toml as toml
|
||||
import salt.serializers.yaml as yaml
|
||||
import salt.serializers.yamlex as yamlex
|
||||
import yaml as _yaml # future lint: disable=blacklisted-import
|
||||
from salt.serializers import SerializationError
|
||||
from salt.serializers.yaml import EncryptedString
|
||||
from salt.utils.odict import OrderedDict
|
||||
from tests.support.helpers import ON_PY35
|
||||
|
||||
SKIP_MESSAGE = "{} is unavailable, have prerequisites been met?"
|
||||
|
||||
|
||||
@pytest.mark.skipif(json.available is False, reason=SKIP_MESSAGE.format("json"))
|
||||
def test_serialize_json():
|
||||
data = {"foo": "bar"}
|
||||
serialized = json.serialize(data)
|
||||
assert serialized == '{"foo": "bar"}', serialized
|
||||
|
||||
deserialized = json.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(yaml.available is False, reason=SKIP_MESSAGE.format("yaml"))
|
||||
def test_serialize_yaml():
|
||||
data = {"foo": "bar", "encrypted_data": EncryptedString("foo")}
|
||||
# The C dumper produces unquoted strings when serializing an
|
||||
# EncryptedString, while the non-C dumper produces quoted strings.
|
||||
expected = (
|
||||
"{encrypted_data: !encrypted foo, foo: bar}"
|
||||
if hasattr(_yaml, "CSafeDumper")
|
||||
else "{encrypted_data: !encrypted 'foo', foo: bar}"
|
||||
)
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == expected, serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(yaml.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_serialize_sls():
|
||||
data = {"foo": "bar"}
|
||||
serialized = yamlex.serialize(data)
|
||||
assert serialized == "{foo: bar}", serialized
|
||||
|
||||
serialized = yamlex.serialize(data, default_flow_style=False)
|
||||
assert serialized == "foo: bar", serialized
|
||||
|
||||
deserialized = yamlex.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == "{foo: bar}", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data, default_flow_style=False)
|
||||
assert serialized == "foo: bar", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_serialize_complex_sls():
|
||||
data = OrderedDict([("foo", 1), ("bar", 2), ("baz", True)])
|
||||
serialized = yamlex.serialize(data)
|
||||
assert serialized == "{foo: 1, bar: 2, baz: true}", serialized
|
||||
|
||||
deserialized = yamlex.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == "{bar: 2, baz: true, foo: 1}", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(yaml.available is False, reason=SKIP_MESSAGE.format("yaml"))
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_compare_sls_vs_yaml():
|
||||
src = "{foo: 1, bar: 2, baz: {qux: true}}"
|
||||
sls_data = yamlex.deserialize(src)
|
||||
yml_data = yaml.deserialize(src)
|
||||
|
||||
# ensure that sls & yaml have the same base
|
||||
assert isinstance(sls_data, dict)
|
||||
assert isinstance(yml_data, dict)
|
||||
assert sls_data == yml_data
|
||||
|
||||
# ensure that sls is ordered, while yaml not
|
||||
assert isinstance(sls_data, OrderedDict)
|
||||
assert not isinstance(yml_data, OrderedDict)
|
||||
|
||||
|
||||
@pytest.mark.skipif(yaml.available is False, reason=SKIP_MESSAGE.format("yaml"))
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
@pytest.mark.skipif(ON_PY35 is True, reason="This test is unreliable under Py3.5")
|
||||
def test_compare_sls_vs_yaml_with_jinja():
|
||||
tpl = "{{ data }}"
|
||||
env = jinja2.Environment()
|
||||
src = "{foo: 1, bar: 2, baz: {qux: true}}"
|
||||
|
||||
sls_src = env.from_string(tpl).render(data=yamlex.deserialize(src))
|
||||
yml_src = env.from_string(tpl).render(data=yaml.deserialize(src))
|
||||
|
||||
sls_data = yamlex.deserialize(sls_src)
|
||||
yml_data = yaml.deserialize(yml_src)
|
||||
|
||||
# ensure that sls & yaml have the same base
|
||||
assert isinstance(sls_data, dict)
|
||||
assert isinstance(yml_data, dict)
|
||||
# The below has been commented out because something the loader test
|
||||
# is modifying the yaml renderer to render things to unicode. Without
|
||||
# running the loader test, the below passes. Even reloading the module
|
||||
# from disk does not reset its internal state (per the Python docs).
|
||||
##
|
||||
# assert sls_data == yml_data
|
||||
|
||||
# ensure that sls is ordered, while yaml not
|
||||
assert isinstance(sls_data, OrderedDict)
|
||||
assert not isinstance(yml_data, OrderedDict)
|
||||
|
||||
# prove that yaml does not handle well with OrderedDict
|
||||
# while sls is jinja friendly.
|
||||
obj = OrderedDict([("foo", 1), ("bar", 2), ("baz", {"qux": True})])
|
||||
|
||||
sls_obj = yamlex.deserialize(yamlex.serialize(obj))
|
||||
try:
|
||||
yml_obj = yaml.deserialize(yaml.serialize(obj))
|
||||
except SerializationError:
|
||||
# BLAAM! yaml was unable to serialize OrderedDict,
|
||||
# but it's not the purpose of the current test.
|
||||
yml_obj = obj.copy()
|
||||
|
||||
sls_src = env.from_string(tpl).render(data=sls_obj)
|
||||
yml_src = env.from_string(tpl).render(data=yml_obj)
|
||||
|
||||
final_obj = yaml.deserialize(sls_src)
|
||||
assert obj == final_obj
|
||||
|
||||
# BLAAM! yml_src is not valid !
|
||||
final_obj = OrderedDict(yaml.deserialize(yml_src))
|
||||
assert obj != final_obj, "Objects matched! {} == {}".format(obj, final_obj)
|
||||
|
||||
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_sls_aggregate():
|
||||
src = dedent(
|
||||
"""
|
||||
a: lol
|
||||
foo: !aggregate hello
|
||||
bar: !aggregate [1, 2, 3]
|
||||
baz: !aggregate
|
||||
a: 42
|
||||
b: 666
|
||||
c: the beast
|
||||
"""
|
||||
).strip()
|
||||
|
||||
# test that !aggregate is correctly parsed
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"a": "lol",
|
||||
"foo": ["hello"],
|
||||
"bar": [1, 2, 3],
|
||||
"baz": {"a": 42, "b": 666, "c": "the beast"},
|
||||
}, sls_obj
|
||||
|
||||
assert (
|
||||
dedent(
|
||||
"""
|
||||
a: lol
|
||||
foo: [hello]
|
||||
bar: [1, 2, 3]
|
||||
baz: {a: 42, b: 666, c: the beast}
|
||||
"""
|
||||
).strip()
|
||||
== yamlex.serialize(sls_obj)
|
||||
), sls_obj
|
||||
|
||||
# test that !aggregate aggregates scalars
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate foo
|
||||
placeholder: !aggregate bar
|
||||
placeholder: !aggregate baz
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": ["foo", "bar", "baz"]}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates lists
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate foo
|
||||
placeholder: !aggregate [bar, baz]
|
||||
placeholder: !aggregate []
|
||||
placeholder: !aggregate ~
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": ["foo", "bar", "baz"]}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates dicts
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate {foo: 42}
|
||||
placeholder: !aggregate {bar: null}
|
||||
placeholder: !aggregate {baz: inga}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": {"foo": 42, "bar": None, "baz": "inga"}}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates deep dicts
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {foo: !aggregate {foo: 42}}
|
||||
placeholder: {foo: !aggregate {bar: null}}
|
||||
placeholder: {foo: !aggregate {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"placeholder": {"foo": {"foo": 42, "bar": None, "baz": "inga"}}
|
||||
}, sls_obj
|
||||
|
||||
# test that {foo: !aggregate bar} and {!aggregate foo: bar}
|
||||
# are roughly equivalent.
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {!aggregate foo: {foo: 42}}
|
||||
placeholder: {!aggregate foo: {bar: null}}
|
||||
placeholder: {!aggregate foo: {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"placeholder": {"foo": {"foo": 42, "bar": None, "baz": "inga"}}
|
||||
}, sls_obj
|
||||
|
||||
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_sls_reset():
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {!aggregate foo: {foo: 42}}
|
||||
placeholder: {!aggregate foo: {bar: null}}
|
||||
!reset placeholder: {!aggregate foo: {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": {"foo": {"baz": "inga"}}}, sls_obj
|
||||
|
||||
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_sls_repr():
|
||||
"""
|
||||
Ensure that obj __repr__ and __str__ methods are yaml friendly.
|
||||
"""
|
||||
|
||||
def convert(obj):
|
||||
return yamlex.deserialize(yamlex.serialize(obj))
|
||||
|
||||
sls_obj = convert(OrderedDict([("foo", "bar"), ("baz", "qux")]))
|
||||
|
||||
# ensure that repr and str are yaml friendly
|
||||
assert sls_obj.__str__() == "{foo: bar, baz: qux}"
|
||||
assert sls_obj.__repr__() == "{foo: bar, baz: qux}"
|
||||
|
||||
# ensure that repr and str are already quoted
|
||||
assert sls_obj["foo"].__str__() == '"bar"'
|
||||
assert sls_obj["foo"].__repr__() == '"bar"'
|
||||
|
||||
|
||||
@pytest.mark.skipif(yamlex.available is False, reason=SKIP_MESSAGE.format("sls"))
|
||||
def test_sls_micking_file_merging():
|
||||
def convert(obj):
|
||||
return yamlex.deserialize(yamlex.serialize(obj))
|
||||
|
||||
# let say that we have 2 pillar files
|
||||
|
||||
src1 = dedent(
|
||||
"""
|
||||
a: first
|
||||
b: !aggregate first
|
||||
c:
|
||||
subkey1: first
|
||||
subkey2: !aggregate first
|
||||
"""
|
||||
).strip()
|
||||
|
||||
src2 = dedent(
|
||||
"""
|
||||
a: second
|
||||
b: !aggregate second
|
||||
c:
|
||||
subkey2: !aggregate second
|
||||
subkey3: second
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj1 = yamlex.deserialize(src1)
|
||||
sls_obj2 = yamlex.deserialize(src2)
|
||||
sls_obj3 = yamlex.merge_recursive(sls_obj1, sls_obj2)
|
||||
|
||||
assert sls_obj3 == {
|
||||
"a": "second",
|
||||
"b": ["first", "second"],
|
||||
"c": {"subkey2": ["first", "second"], "subkey3": "second"},
|
||||
}, sls_obj3
|
||||
|
||||
|
||||
@pytest.mark.skipif(msgpack.available is False, reason=SKIP_MESSAGE.format("msgpack"))
|
||||
def test_msgpack():
|
||||
data = OrderedDict([("foo", 1), ("bar", 2), ("baz", True)])
|
||||
serialized = msgpack.serialize(data)
|
||||
deserialized = msgpack.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(python.available is False, reason=SKIP_MESSAGE.format("python"))
|
||||
def test_serialize_python():
|
||||
data = {"foo": "bar"}
|
||||
serialized = python.serialize(data)
|
||||
expected = repr({"foo": "bar"})
|
||||
assert serialized == expected, serialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
configparser.available is False, reason=SKIP_MESSAGE.format("configparser")
|
||||
)
|
||||
def test_configparser():
|
||||
data = {"foo": {"bar": "baz"}}
|
||||
# configparser appends empty lines
|
||||
serialized = configparser.serialize(data).strip()
|
||||
assert serialized == "[foo]\nbar = baz", serialized
|
||||
|
||||
deserialized = configparser.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(toml.available is False, reason=SKIP_MESSAGE.format("toml"))
|
||||
def test_serialize_toml():
|
||||
data = {"foo": "bar"}
|
||||
serialized = toml.serialize(data)
|
||||
assert serialized == 'foo = "bar"\n', serialized
|
||||
|
||||
deserialized = toml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(plist.available is False, reason=SKIP_MESSAGE.format("plist"))
|
||||
def test_serialize_plist():
|
||||
data = {"foo": "bar"}
|
||||
serialized = plist.serialize(data)
|
||||
expected = (
|
||||
b'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
b'<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n'
|
||||
b'<plist version="1.0">\n'
|
||||
b"<dict>\n"
|
||||
b"\t<key>foo</key>\n"
|
||||
b"\t<string>bar</string>\n"
|
||||
b"</dict>\n"
|
||||
b"</plist>\n"
|
||||
)
|
||||
assert serialized == expected, serialized
|
||||
|
||||
deserialized = plist.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
|
||||
@pytest.mark.skipif(plist.available is False, reason=SKIP_MESSAGE.format("plist"))
|
||||
def test_serialize_binary_plist():
|
||||
data = {"foo": "bar"}
|
||||
serialized = plist.serialize(data, fmt="FMT_BINARY")
|
||||
|
||||
deserialized = plist.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
|
@ -62,6 +62,7 @@ PRE_PYTEST_SKIP = pytest.mark.skipif(
|
|||
SKIP_IF_NOT_RUNNING_PYTEST = skipIf(
|
||||
RUNTIME_VARS.PYTEST_SESSION is False, "These tests now require running under PyTest"
|
||||
)
|
||||
ON_PY35 = sys.version_info < (3, 6)
|
||||
|
||||
|
||||
def no_symlinks():
|
||||
|
|
|
@ -1 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
|
@ -1,384 +0,0 @@
|
|||
from textwrap import dedent
|
||||
|
||||
import jinja2
|
||||
import salt.serializers.configparser as configparser
|
||||
import salt.serializers.json as json
|
||||
import salt.serializers.msgpack as msgpack
|
||||
import salt.serializers.plist as plist
|
||||
import salt.serializers.python as python
|
||||
import salt.serializers.toml as toml
|
||||
import salt.serializers.yaml as yaml
|
||||
import salt.serializers.yamlex as yamlex
|
||||
import yaml as _yaml # future lint: disable=blacklisted-import
|
||||
from salt.serializers import SerializationError
|
||||
from salt.serializers.yaml import EncryptedString
|
||||
from salt.utils.odict import OrderedDict
|
||||
from tests.support.helpers import flaky
|
||||
from tests.support.unit import TestCase, skipIf
|
||||
|
||||
SKIP_MESSAGE = "%s is unavailable, have prerequisites been met?"
|
||||
|
||||
|
||||
@flaky(condition=True)
|
||||
class TestSerializers(TestCase):
|
||||
@skipIf(not json.available, SKIP_MESSAGE % "json")
|
||||
def test_serialize_json(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = json.serialize(data)
|
||||
assert serialized == '{"foo": "bar"}', serialized
|
||||
|
||||
deserialized = json.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not yaml.available, SKIP_MESSAGE % "yaml")
|
||||
def test_serialize_yaml(self):
|
||||
data = {"foo": "bar", "encrypted_data": EncryptedString("foo")}
|
||||
# The C dumper produces unquoted strings when serializing an
|
||||
# EncryptedString, while the non-C dumper produces quoted strings.
|
||||
expected = (
|
||||
"{encrypted_data: !encrypted foo, foo: bar}"
|
||||
if hasattr(_yaml, "CSafeDumper")
|
||||
else "{encrypted_data: !encrypted 'foo', foo: bar}"
|
||||
)
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == expected, serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not yaml.available, SKIP_MESSAGE % "sls")
|
||||
def test_serialize_sls(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = yamlex.serialize(data)
|
||||
assert serialized == "{foo: bar}", serialized
|
||||
|
||||
serialized = yamlex.serialize(data, default_flow_style=False)
|
||||
assert serialized == "foo: bar", serialized
|
||||
|
||||
deserialized = yamlex.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == "{foo: bar}", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data, default_flow_style=False)
|
||||
assert serialized == "foo: bar", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_serialize_complex_sls(self):
|
||||
data = OrderedDict([("foo", 1), ("bar", 2), ("baz", True)])
|
||||
serialized = yamlex.serialize(data)
|
||||
assert serialized == "{foo: 1, bar: 2, baz: true}", serialized
|
||||
|
||||
deserialized = yamlex.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
serialized = yaml.serialize(data)
|
||||
assert serialized == "{bar: 2, baz: true, foo: 1}", serialized
|
||||
|
||||
deserialized = yaml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not yaml.available, SKIP_MESSAGE % "yaml")
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_compare_sls_vs_yaml(self):
|
||||
src = "{foo: 1, bar: 2, baz: {qux: true}}"
|
||||
sls_data = yamlex.deserialize(src)
|
||||
yml_data = yaml.deserialize(src)
|
||||
|
||||
# ensure that sls & yaml have the same base
|
||||
assert isinstance(sls_data, dict)
|
||||
assert isinstance(yml_data, dict)
|
||||
assert sls_data == yml_data
|
||||
|
||||
# ensure that sls is ordered, while yaml not
|
||||
assert isinstance(sls_data, OrderedDict)
|
||||
assert not isinstance(yml_data, OrderedDict)
|
||||
|
||||
@skipIf(not yaml.available, SKIP_MESSAGE % "yaml")
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
@skipIf(True, "Flaky on Python 3.")
|
||||
def test_compare_sls_vs_yaml_with_jinja(self):
|
||||
tpl = "{{ data }}"
|
||||
env = jinja2.Environment()
|
||||
src = "{foo: 1, bar: 2, baz: {qux: true}}"
|
||||
|
||||
sls_src = env.from_string(tpl).render(data=yamlex.deserialize(src))
|
||||
yml_src = env.from_string(tpl).render(data=yaml.deserialize(src))
|
||||
|
||||
sls_data = yamlex.deserialize(sls_src)
|
||||
yml_data = yaml.deserialize(yml_src)
|
||||
|
||||
# ensure that sls & yaml have the same base
|
||||
assert isinstance(sls_data, dict)
|
||||
assert isinstance(yml_data, dict)
|
||||
# The below has been commented out because something the loader test
|
||||
# is modifying the yaml renderer to render things to unicode. Without
|
||||
# running the loader test, the below passes. Even reloading the module
|
||||
# from disk does not reset its internal state (per the Python docs).
|
||||
##
|
||||
# assert sls_data == yml_data
|
||||
|
||||
# ensure that sls is ordered, while yaml not
|
||||
assert isinstance(sls_data, OrderedDict)
|
||||
assert not isinstance(yml_data, OrderedDict)
|
||||
|
||||
# prove that yaml does not handle well with OrderedDict
|
||||
# while sls is jinja friendly.
|
||||
obj = OrderedDict([("foo", 1), ("bar", 2), ("baz", {"qux": True})])
|
||||
|
||||
sls_obj = yamlex.deserialize(yamlex.serialize(obj))
|
||||
try:
|
||||
yml_obj = yaml.deserialize(yaml.serialize(obj))
|
||||
except SerializationError:
|
||||
# BLAAM! yaml was unable to serialize OrderedDict,
|
||||
# but it's not the purpose of the current test.
|
||||
yml_obj = obj.copy()
|
||||
|
||||
sls_src = env.from_string(tpl).render(data=sls_obj)
|
||||
yml_src = env.from_string(tpl).render(data=yml_obj)
|
||||
|
||||
final_obj = yaml.deserialize(sls_src)
|
||||
assert obj == final_obj
|
||||
|
||||
# BLAAM! yml_src is not valid !
|
||||
final_obj = OrderedDict(yaml.deserialize(yml_src))
|
||||
assert obj != final_obj, "Objects matched! {} == {}".format(obj, final_obj)
|
||||
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_sls_aggregate(self):
|
||||
src = dedent(
|
||||
"""
|
||||
a: lol
|
||||
foo: !aggregate hello
|
||||
bar: !aggregate [1, 2, 3]
|
||||
baz: !aggregate
|
||||
a: 42
|
||||
b: 666
|
||||
c: the beast
|
||||
"""
|
||||
).strip()
|
||||
|
||||
# test that !aggregate is correctly parsed
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"a": "lol",
|
||||
"foo": ["hello"],
|
||||
"bar": [1, 2, 3],
|
||||
"baz": {"a": 42, "b": 666, "c": "the beast"},
|
||||
}, sls_obj
|
||||
|
||||
assert (
|
||||
dedent(
|
||||
"""
|
||||
a: lol
|
||||
foo: [hello]
|
||||
bar: [1, 2, 3]
|
||||
baz: {a: 42, b: 666, c: the beast}
|
||||
"""
|
||||
).strip()
|
||||
== yamlex.serialize(sls_obj)
|
||||
), sls_obj
|
||||
|
||||
# test that !aggregate aggregates scalars
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate foo
|
||||
placeholder: !aggregate bar
|
||||
placeholder: !aggregate baz
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": ["foo", "bar", "baz"]}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates lists
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate foo
|
||||
placeholder: !aggregate [bar, baz]
|
||||
placeholder: !aggregate []
|
||||
placeholder: !aggregate ~
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": ["foo", "bar", "baz"]}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates dicts
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: !aggregate {foo: 42}
|
||||
placeholder: !aggregate {bar: null}
|
||||
placeholder: !aggregate {baz: inga}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"placeholder": {"foo": 42, "bar": None, "baz": "inga"}
|
||||
}, sls_obj
|
||||
|
||||
# test that !aggregate aggregates deep dicts
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {foo: !aggregate {foo: 42}}
|
||||
placeholder: {foo: !aggregate {bar: null}}
|
||||
placeholder: {foo: !aggregate {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"placeholder": {"foo": {"foo": 42, "bar": None, "baz": "inga"}}
|
||||
}, sls_obj
|
||||
|
||||
# test that {foo: !aggregate bar} and {!aggregate foo: bar}
|
||||
# are roughly equivalent.
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {!aggregate foo: {foo: 42}}
|
||||
placeholder: {!aggregate foo: {bar: null}}
|
||||
placeholder: {!aggregate foo: {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {
|
||||
"placeholder": {"foo": {"foo": 42, "bar": None, "baz": "inga"}}
|
||||
}, sls_obj
|
||||
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_sls_reset(self):
|
||||
src = dedent(
|
||||
"""
|
||||
placeholder: {!aggregate foo: {foo: 42}}
|
||||
placeholder: {!aggregate foo: {bar: null}}
|
||||
!reset placeholder: {!aggregate foo: {baz: inga}}
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj = yamlex.deserialize(src)
|
||||
assert sls_obj == {"placeholder": {"foo": {"baz": "inga"}}}, sls_obj
|
||||
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_sls_repr(self):
|
||||
"""
|
||||
Ensure that obj __repr__ and __str__ methods are yaml friendly.
|
||||
"""
|
||||
|
||||
def convert(obj):
|
||||
return yamlex.deserialize(yamlex.serialize(obj))
|
||||
|
||||
sls_obj = convert(OrderedDict([("foo", "bar"), ("baz", "qux")]))
|
||||
|
||||
# ensure that repr and str are yaml friendly
|
||||
assert sls_obj.__str__() == "{foo: bar, baz: qux}"
|
||||
assert sls_obj.__repr__() == "{foo: bar, baz: qux}"
|
||||
|
||||
# ensure that repr and str are already quoted
|
||||
assert sls_obj["foo"].__str__() == '"bar"'
|
||||
assert sls_obj["foo"].__repr__() == '"bar"'
|
||||
|
||||
@skipIf(not yamlex.available, SKIP_MESSAGE % "sls")
|
||||
def test_sls_micking_file_merging(self):
|
||||
def convert(obj):
|
||||
return yamlex.deserialize(yamlex.serialize(obj))
|
||||
|
||||
# let say that we have 2 pillar files
|
||||
|
||||
src1 = dedent(
|
||||
"""
|
||||
a: first
|
||||
b: !aggregate first
|
||||
c:
|
||||
subkey1: first
|
||||
subkey2: !aggregate first
|
||||
"""
|
||||
).strip()
|
||||
|
||||
src2 = dedent(
|
||||
"""
|
||||
a: second
|
||||
b: !aggregate second
|
||||
c:
|
||||
subkey2: !aggregate second
|
||||
subkey3: second
|
||||
"""
|
||||
).strip()
|
||||
|
||||
sls_obj1 = yamlex.deserialize(src1)
|
||||
sls_obj2 = yamlex.deserialize(src2)
|
||||
sls_obj3 = yamlex.merge_recursive(sls_obj1, sls_obj2)
|
||||
|
||||
assert sls_obj3 == {
|
||||
"a": "second",
|
||||
"b": ["first", "second"],
|
||||
"c": {"subkey2": ["first", "second"], "subkey3": "second"},
|
||||
}, sls_obj3
|
||||
|
||||
@skipIf(not msgpack.available, SKIP_MESSAGE % "msgpack")
|
||||
def test_msgpack(self):
|
||||
data = OrderedDict([("foo", 1), ("bar", 2), ("baz", True)])
|
||||
serialized = msgpack.serialize(data)
|
||||
deserialized = msgpack.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not python.available, SKIP_MESSAGE % "python")
|
||||
def test_serialize_python(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = python.serialize(data)
|
||||
expected = repr({"foo": "bar"})
|
||||
assert serialized == expected, serialized
|
||||
|
||||
@skipIf(not configparser.available, SKIP_MESSAGE % "configparser")
|
||||
def test_configparser(self):
|
||||
data = {"foo": {"bar": "baz"}}
|
||||
# configparser appends empty lines
|
||||
serialized = configparser.serialize(data).strip()
|
||||
assert serialized == "[foo]\nbar = baz", serialized
|
||||
|
||||
deserialized = configparser.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not toml.available, SKIP_MESSAGE % "toml")
|
||||
def test_serialize_toml(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = toml.serialize(data)
|
||||
assert serialized == 'foo = "bar"\n', serialized
|
||||
|
||||
deserialized = toml.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not plist.available, SKIP_MESSAGE % "plist")
|
||||
def test_serialize_plist(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = plist.serialize(data)
|
||||
expected = (
|
||||
b'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
b'<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n'
|
||||
b'<plist version="1.0">\n'
|
||||
b"<dict>\n"
|
||||
b"\t<key>foo</key>\n"
|
||||
b"\t<string>bar</string>\n"
|
||||
b"</dict>\n"
|
||||
b"</plist>\n"
|
||||
)
|
||||
assert serialized == expected, serialized
|
||||
|
||||
deserialized = plist.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
||||
|
||||
@skipIf(not plist.available, SKIP_MESSAGE % "plist")
|
||||
def test_serialize_binary_plist(self):
|
||||
data = {"foo": "bar"}
|
||||
serialized = plist.serialize(data, fmt="FMT_BINARY")
|
||||
|
||||
deserialized = plist.deserialize(serialized)
|
||||
assert deserialized == data, deserialized
|
Loading…
Add table
Reference in a new issue