fix parallel mode py3 compatibility

parallel: True codepath incompatibilities uncovered by the added tests.
additionally use salt.serializers.msgpack to avoid other py2/py3/back compat
issues.

salt/state.py

@@ -40,6 +40,7 @@ import salt.utils.process
 import salt.utils.files
 import salt.syspaths as syspaths
 from salt.utils import immutabletypes
+from salt.serializers.msgpack import serialize as msgpack_serialize, deserialize as msgpack_deserialize
 from salt.template import compile_template, compile_template_str
 from salt.exceptions import (
     SaltException,
@@ -56,7 +57,6 @@ import salt.utils.yamlloader as yamlloader
 import salt.ext.six as six
 from salt.ext.six.moves import map, range, reload_module
 # pylint: enable=import-error,no-name-in-module,redefined-builtin
-import msgpack
 
 log = logging.getLogger(__name__)
 
@@ -1731,7 +1731,7 @@ class State(object):
                 # and the attempt, we are safe to pass
                 pass
         with salt.utils.fopen(tfile, 'wb+') as fp_:
-            fp_.write(msgpack.dumps(ret))
+            fp_.write(msgpack_serialize(ret))
 
     def call_parallel(self, cdata, low):
         '''
@@ -2094,7 +2094,7 @@ class State(object):
                                'changes': {}}
                     try:
                         with salt.utils.fopen(ret_cache, 'rb') as fp_:
-                            ret = msgpack.loads(fp_.read())
+                            ret = msgpack_deserialize(fp_.read())
                     except (OSError, IOError):
                         ret = {'result': False,
                                'comment': 'Parallel cache failure',

salt/utils/files.py

@@ -11,7 +11,6 @@ import re
 import shutil
 import subprocess
 import time
-import urllib
 
 # Import salt libs
 import salt.utils
@@ -21,6 +20,12 @@ from salt.exceptions import CommandExecutionError, FileLockError, MinionError
 # Import 3rd-party libs
 from salt.ext import six
 
+if six.PY2:
+    from urllib import quote
+elif six.PY3:
+    from urllib.parse import quote  # pylint: disable=no-name-in-module
+
+
 log = logging.getLogger(__name__)
 
 LOCAL_PROTOS = ('', 'file')
@@ -312,7 +317,7 @@ def safe_filename_leaf(file_basename):
     :codeauthor: Damon Atkins <https://github.com/damon-atkins>
     '''
     def _replace(re_obj):
-        return urllib.quote(re_obj.group(0), safe=u'')
+        return quote(re_obj.group(0), safe=u'')
     if not isinstance(file_basename, six.text_type):
         # the following string is not prefixed with u
         return re.sub('[\\\\:/*?"<>|]',

tests/integration/runners/test_state.py

@@ -334,6 +334,7 @@ class OrchEventTest(ShellCase):
 
         signal.signal(signal.SIGALRM, self.alarm_handler)
         signal.alarm(self.timeout)
+        received = False
         try:
             while True:
                 event = listener.get_event(full=True)
@@ -344,6 +345,7 @@ class OrchEventTest(ShellCase):
                 # the test is implicitly sucessful; if it were happening in serial it would be
                 # atleast 110 seconds.
                 if event['tag'] == 'salt/run/{0}/ret'.format(jid):
+                    received = True
                     # Don't wrap this in a try/except. We want to know if the
                     # data structure is different from what we expect!
                     ret = event['data']['return']['data']['master']
@@ -356,5 +358,6 @@ class OrchEventTest(ShellCase):
                 # self confirm that the total runtime is roughly 30s (left 10s for buffer)
                 self.assertTrue((time.time() - start_time) < 40)
         finally:
+            self.assertTrue(received)
             del listener
             signal.alarm(0)