text: # -*- coding: utf-8 -*-
  
  # Import python libs
  from __future__ import absolute_import, print_function, unicode_literals
  from textwrap import dedent
  
  # Import Salt Testing libs
  from tests.support.unit import skipIf, TestCase
  
  # Import 3rd party libs
  import jinja2
  import yaml as _yaml  # future lint: disable=blacklisted-import
  from salt.ext import six
  
  # Import salt libs
  import salt.serializers.configparser as configparser
  import salt.serializers.json as json
  import salt.serializers.yaml as yaml
  import salt.serializers.yamlex as yamlex
  import salt.serializers.msgpack as msgpack
  import salt.serializers.python as python
  import salt.serializers.toml as toml
  from salt.serializers.yaml import EncryptedString
  from salt.serializers import SerializationError
  from salt.utils.odict import OrderedDict
  
  # Import test support libs
  from tests.support.helpers import flaky
  
  SKIP_MESSAGE = '%s is unavailable, have prerequisites been met?'
  
  
  @flaky(condition=six.PY3)
  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
  
          deserialized = yamlex.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
  
      @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(six.PY3, '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
  
      @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
  