2013-08-01 00:34:37 +00:00
|
|
|
# -*- coding: utf-8 -*-
|
|
|
|
'''
|
|
|
|
:codeauthor: :email:`Pedro Algarvio (pedro@algarvio.me)`
|
2013-09-16 16:24:00 +00:00
|
|
|
:copyright: © 2013 by the SaltStack Team, see AUTHORS for more details
|
2013-08-01 00:34:37 +00:00
|
|
|
:license: Apache 2.0, see LICENSE for more details.
|
2013-09-16 16:24:00 +00:00
|
|
|
|
|
|
|
|
|
|
|
tests.unit.payload_test
|
|
|
|
~~~~~~~~~~~~~~~~~~~~~~~
|
2013-08-01 00:34:37 +00:00
|
|
|
'''
|
|
|
|
|
|
|
|
# Import Salt Testing libs
|
|
|
|
from salttesting import skipIf, TestCase
|
|
|
|
from salttesting.helpers import ensure_in_syspath, MockWraps
|
2013-08-26 10:03:19 +00:00
|
|
|
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch
|
2013-08-01 00:34:37 +00:00
|
|
|
ensure_in_syspath('../')
|
|
|
|
|
|
|
|
# Import salt libs
|
|
|
|
import salt.payload
|
|
|
|
from salt.utils.odict import OrderedDict
|
|
|
|
|
|
|
|
# Import 3rd-party libs
|
|
|
|
import msgpack
|
|
|
|
|
|
|
|
|
2013-08-26 10:03:19 +00:00
|
|
|
@skipIf(NO_MOCK, NO_MOCK_REASON)
|
2013-08-01 00:34:37 +00:00
|
|
|
class PayloadTestCase(TestCase):
|
|
|
|
|
|
|
|
def assertNoOrderedDict(self, data):
|
|
|
|
if isinstance(data, OrderedDict):
|
|
|
|
raise AssertionError(
|
|
|
|
'Found an ordered dictionary'
|
|
|
|
)
|
|
|
|
if isinstance(data, dict):
|
|
|
|
for value in data.values():
|
|
|
|
self.assertNoOrderedDict(value)
|
|
|
|
elif isinstance(data, (list, tuple)):
|
|
|
|
for chunk in data:
|
|
|
|
self.assertNoOrderedDict(chunk)
|
|
|
|
|
|
|
|
def test_list_nested_odicts(self):
|
|
|
|
with patch('msgpack.version', (0, 1, 13)):
|
|
|
|
msgpack.dumps = MockWraps(
|
|
|
|
msgpack.dumps, 1, TypeError('ODict TypeError Forced')
|
|
|
|
)
|
|
|
|
payload = salt.payload.Serial('msgpack')
|
|
|
|
idata = {'pillar': [OrderedDict(environment='dev')]}
|
|
|
|
odata = payload.loads(payload.dumps(idata.copy()))
|
|
|
|
self.assertNoOrderedDict(odata)
|
|
|
|
self.assertEqual(idata, odata)
|
2014-03-31 02:00:24 +00:00
|
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
from integration import run_tests
|
|
|
|
run_tests(PayloadTestCase, needs_daemon=False)
|