Repository: acorg/slurm-pipeline

Issue:
Add a way to have an error step

Discussion:

Patch:
diff --git a/README.md b/README.md
index 33c0168..307f04c 100644
--- a/README.md
+++ b/README.md
@@ -199,6 +199,8 @@ the full list:
 * `collect`: for scripts that should run only when all tasks from all their
   prerequisites have completed.
 * `dependencies`: a list of previous steps that a step depends on.
+* `error step`: if `true` the step script will only be run if one of its
+  dependencies fails.
 * `skip`: if `true`, the step script will be run with `SP_SKIP=1` in its
   environment. Otherwise, `SP_SKIP` will always be set and will be `0`.
 
diff --git a/setup.py b/setup.py
index 7971140..2071b15 100644
--- a/setup.py
+++ b/setup.py
@@ -3,7 +3,7 @@
 from setuptools import setup
 
 setup(name='slurm-pipeline',
-      version='1.0.13',
+      version='1.0.14',
       packages=['slurm_pipeline'],
       include_package_data=True,
       url='https://github.com/acorg/slurm-pipeline',
diff --git a/slurm_pipeline/pipeline.py b/slurm_pipeline/pipeline.py
index 713272d..8161b2c 100644
--- a/slurm_pipeline/pipeline.py
+++ b/slurm_pipeline/pipeline.py
@@ -159,6 +159,13 @@ class SlurmPipeline(object):
         step['skip'] = skip
         scriptArgsStr = ' '.join(map(str, scriptArgs)) if scriptArgs else ''
 
+        if step.get('error step', False):
+            separator = '?'
+            after = 'afternotok'
+        else:
+            separator = ','
+            after = 'afterok'
+
         # taskDependencies is keyed by task name. These are the tasks
         # started by the steps that the current step depends on.  Its
         # values are sets of SLURM job ids the tasks that step started and
@@ -178,8 +185,8 @@ class SlurmPipeline(object):
                 env['SP_ORIGINAL_ARGS'] = scriptArgsStr
                 env['SP_SIMULATE'] = str(int(simulate))
                 env['SP_SKIP'] = str(int(skip))
-                dependencies = ','.join(
-                    sorted(('afterok:%d' % jobId)
+                dependencies = separator.join(
+                    sorted(('%s:%d' % (after, jobId))
                            for jobIds in taskDependencies.values()
                            for jobId in jobIds))
                 if dependencies:
@@ -196,8 +203,9 @@ class SlurmPipeline(object):
                     env['SP_SIMULATE'] = str(int(simulate))
                     env['SP_SKIP'] = str(int(skip))
                     jobIds = steps[stepName]['tasks'][taskName]
-                    dependencies = ','.join(sorted(('afterok:%d' % jobId)
-                                                   for jobId in jobIds))
+                    dependencies = separator.join(
+                        sorted(('%s:%d' % (after, jobId))
+                               for jobId in jobIds))
                     if dependencies:
                         env['SP_DEPENDENCY_ARG'] = ('--dependency=' +
                                                     dependencies)
@@ -352,6 +360,11 @@ class SlurmPipeline(object):
                             'Step %d depends on a non-existent (or '
                             'not-yet-defined) step: %r' % (count, dependency))
 
+            if 'error step' in step and 'dependencies' not in step:
+                    raise SpecificationError(
+                        'Step "%s" is an error step but has no "dependencies" '
+                        'key' % (step['name']))
+
     def _checkRuntime(self, steps, firstStep=None, lastStep=None, skip=None):
         """
         Check that a proposed scheduling makes sense.


Test patch:
diff --git a/test/test_pipeline.py b/test/test_pipeline.py
index 4e0e396..0057a04 100644
--- a/test/test_pipeline.py
+++ b/test/test_pipeline.py
@@ -1491,3 +1491,76 @@ class TestRunner(TestCase):
             },
             sort_keys=True, indent=2, separators=(',', ': '))
         self.assertEqual(expected, runner.specificationToJSON(specification))
+
+    @patch('os.access')
+    @patch('os.path.exists')
+    def testErrorStepWithNoDependencies(self, existsMock, accessMock):
+        """
+        If the specification steps contains an error step that does not have
+        any dependencies, a SpecificationError must be raised because error
+        steps only exist for the purposed of catching failed dependencies.
+        """
+        error = '^Step "xx" is an error step but has no "dependencies" key$'
+        assertRaisesRegex(self, SpecificationError, error, SlurmPipeline,
+                          {
+                              'steps': [
+                                  {
+                                      'error step': True,
+                                      'name': 'xx',
+                                      'script': 'script',
+                                  },
+                              ]
+                          })
+
+    @patch('subprocess.check_output')
+    @patch('os.access')
+    @patch('os.path.exists')
+    def testErrorStep(self, existsMock, accessMock, subprocessMock):
+        """
+        If a step is an error step its script must be run with the expected
+        value of SP_DEPENDENCY_ARG.
+        """
+
+        class SideEffect(object):
+            def __init__(self):
+                self.first = True
+
+            def sideEffect(self, *args, **kwargs):
+                if self.first:
+                    self.first = False
+                    return ('TASK: aaa 127 450\n'
+                            'TASK: bbb 238 560\n')
+                else:
+                    return ''
+
+        subprocessMock.side_effect = SideEffect().sideEffect
+
+        runner = SlurmPipeline(
+            {
+                'steps': [
+                    {
+                        'name': 'name1',
+                        'script': 'script1',
+                    },
+                    {
+                        'dependencies': ['name1'],
+                        'error step': True,
+                        'name': 'name2',
+                        'script': 'script2',
+                    },
+                ],
+            })
+        runner.schedule()
+
+        # Check that the dependency environment variable is correct in
+        # all calls.
+        env1 = subprocessMock.mock_calls[0][2]['env']
+        self.assertNotIn('SP_DEPENDENCY_ARG', env1)
+
+        env2 = subprocessMock.mock_calls[1][2]['env']
+        self.assertEqual('--dependency=afternotok:127?afternotok:450',
+                         env2['SP_DEPENDENCY_ARG'])
+
+        env3 = subprocessMock.mock_calls[2][2]['env']
+        self.assertEqual('--dependency=afternotok:238?afternotok:560',
+                         env3['SP_DEPENDENCY_ARG'])