add integration test to runners/test_state to exercise parallel

this should hopefully exercise a few different facets of parallel that were
previously not covered in the code base.

salt/state.py

@@ -1721,7 +1721,6 @@ class State(object):
         duration = (delta.seconds * 1000000 + delta.microseconds)/1000.0
         ret['duration'] = duration
 
-
         troot = os.path.join(self.opts['cachedir'], self.jid)
         tfile = os.path.join(troot, _clean_tag(tag))
         if not os.path.isdir(troot):
@@ -1743,7 +1742,7 @@ class State(object):
         # enough to not raise another KeyError as the name is easily
         # guessable and fallback in all cases to present the real
         # exception to the user
-        if len(cdata['args']) > 0:
+        if 'args' in cdata and cdata['args']:
             name = cdata['args'][0]
         elif 'name' in cdata['kwargs']:
             name = cdata['kwargs']['name']
@@ -1892,7 +1891,7 @@ class State(object):
             # enough to not raise another KeyError as the name is easily
             # guessable and fallback in all cases to present the real
             # exception to the user
-            if len(cdata['args']) > 0:
+            if 'args' in cdata and cdata['args']:
                 name = cdata['args'][0]
             elif 'name' in cdata['kwargs']:
                 name = cdata['kwargs']['name']

tests/integration/runners/test_state.py

@@ -6,10 +6,12 @@ Tests for the state runner
 # Import Python Libs
 from __future__ import absolute_import
 import errno
+import logging
 import os
 import shutil
 import signal
 import tempfile
+import time
 import textwrap
 import yaml
 import threading
@@ -24,6 +26,8 @@ from tests.support.paths import TMP
 import salt.utils
 import salt.utils.event
 
+log = logging.getLogger(__name__)
+
 
 class StateRunnerTest(ShellCase):
     '''
@@ -275,3 +279,90 @@ class OrchEventTest(ShellCase):
         finally:
             del listener
             signal.alarm(0)
+
+    def test_parallel_orchestrations(self):
+        '''
+        Test to confirm that the parallel state requisite works in orch
+        we do this by running 10 test.sleep's of 10 seconds, and insure it only takes roughly 10s
+        '''
+        self.write_conf({
+            'fileserver_backend': ['roots'],
+            'file_roots': {
+                'base': [self.base_env],
+            },
+        })
+
+        orch_sls = os.path.join(self.base_env, 'test_par_orch.sls')
+
+        with salt.utils.fopen(orch_sls, 'w') as fp_:
+            fp_.write(textwrap.dedent('''
+                {% for count in range(1, 20) %}
+
+                sleep {{ count }}:
+                    module.run:
+                        - name: test.sleep
+                        - length: 10
+                        - parallel: True
+
+                {% endfor %}
+
+                sleep 21:
+                    module.run:
+                        - name: test.sleep
+                        - length: 10
+                        - parallel: True
+                        - prereq:
+                            - module: sleep 1
+
+                sleep 22:
+                    module.run:
+                        - name: test.sleep
+                        - length: 10
+                        - parallel: True
+                        - require:
+                            - module: sleep 1
+            '''))
+
+        orch_sls = os.path.join(self.base_env, 'test_par_orch.sls')
+
+        listener = salt.utils.event.get_event(
+            'master',
+            sock_dir=self.master_opts['sock_dir'],
+            transport=self.master_opts['transport'],
+            opts=self.master_opts)
+
+        start_time = time.time()
+        jid = self.run_run_plus(
+            'state.orchestrate',
+            'test_par_orch',
+            __reload_config=True).get('jid')
+
+        if jid is None:
+            raise Exception('jid missing from run_run_plus output')
+
+        signal.signal(signal.SIGALRM, self.alarm_handler)
+        signal.alarm(self.timeout)
+        try:
+            while True:
+                event = listener.get_event(full=True)
+                if event is None:
+                    continue
+
+                # if we receive the ret for this job before self.timeout (60),
+                # 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):
+                    # 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']
+                    for state in ret:
+                        data = ret[state]
+                        # we expect each duration to be greater than 10s
+                        self.assertTrue(data['duration'] > 10000)
+                    break
+
+                # self confirm that the total runtime is roughly 30s (left 10s for buffer)
+                self.assertTrue((time.time() - start_time) < 40)
+        finally:
+            del listener
+            signal.alarm(0)