Add support for exception during Pipeline execution #82

Signed-off-by: Thomas Druez <[email protected]>

scanpipe/pipelines/__init__.py

@@ -23,6 +23,7 @@
 import importlib
 import inspect
 import logging
+import traceback
 from contextlib import contextmanager
 from pydoc import getdoc
 
@@ -66,14 +67,20 @@ def execute(self):
  self.log(f"Pipeline [{self.pipeline_name}] starting")
 
  for step in self.steps:
- step_name = step.__name__
- self.log(f"Step [{step_name}] starting")
- step(self)
- self.log(f"Step [{step_name}] completed")
+ self.log(f"Step [{step.__name__}] starting")
 
- self.log(f"Pipeline [{self.pipeline_name}] completed")
+ try:
+ step(self)
+ except Exception as e:
+ self.log(f"Pipeline failed")
+ tb = "".join(traceback.format_tb(e.__traceback__))
+ return 1, f"{e}\n\nTraceback:\n{tb}"
 
- return 0, "Completed"
+ self.log(f"Step [{step.__name__}] completed")
+
+ self.log(f"Pipeline completed")
+
+ return 0, ""
 
  @contextmanager
  def save_errors(self, *exceptions):

scanpipe/tests/pipelines/do_nothing.py

@@ -0,0 +1,40 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https:/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https:/nexB/scancode.io for support and download.
+
+from scanpipe.pipelines import Pipeline
+
+
+class DoNothing(Pipeline):
+ """
+ A pipeline that does nothing, in 2 steps.
+ """
+
+ def step1(self):
+ pass
+
+ def step2(self):
+ pass
+
+ steps = (
+ step1,
+ step2,
+ )

scanpipe/tests/pipelines/raise_exception.py

@@ -0,0 +1,34 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https:/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https:/nexB/scancode.io for support and download.
+
+from scanpipe.pipelines import Pipeline
+
+
+class RaiseException(Pipeline):
+ """
+ A pipeline that raises an Exception.
+ """
+
+ def raise_exception(self):
+ raise ValueError("Error message")
+
+ steps = (raise_exception,)

scanpipe/tests/test_pipelines.py

@@ -59,6 +59,42 @@ def test_scanpipe_pipeline_class_log(self):
  self.assertIn("Event1", run.log)
  self.assertIn("Event2", run.log)
 
+ def test_scanpipe_pipeline_class_execute(self):
+ project1 = Project.objects.create(name="Analysis")
+ run = project1.add_pipeline("scanpipe/tests/pipelines/do_nothing.py")
+ pipeline = run.make_pipeline_instance()
+
+ exitcode, output = pipeline.execute()
+ self.assertEqual(0, exitcode)
+ self.assertEqual("", output)
+
+ run.refresh_from_db()
+ self.assertIn("Pipeline [DoNothing] starting", run.log)
+ self.assertIn("Step [step1] starting", run.log)
+ self.assertIn("Step [step1] completed", run.log)
+ self.assertIn("Step [step2] starting", run.log)
+ self.assertIn("Step [step2] completed", run.log)
+ self.assertIn("Pipeline completed", run.log)
+
+ def test_scanpipe_pipeline_class_execute_with_exception(self):
+ project1 = Project.objects.create(name="Analysis")
+ run = project1.add_pipeline("scanpipe/tests/pipelines/raise_exception.py")
+ pipeline = run.make_pipeline_instance()
+
+ exitcode, output = pipeline.execute()
+ self.assertEqual(1, exitcode)
+ self.assertTrue(output.startswith("Error message"))
+ self.assertIn("Traceback:", output)
+ self.assertIn("in execute", output)
+ self.assertIn("step(self)", output)
+ self.assertIn("in raise_exception", output)
+ self.assertIn("raise ValueError", output)
+
+ run.refresh_from_db()
+ self.assertIn("Pipeline [RaiseException] starting", run.log)
+ self.assertIn("Step [raise_exception] starting", run.log)
+ self.assertIn("Pipeline failed", run.log)
+
  def test_scanpipe_pipeline_class_save_errors_context_manager(self):
  project1 = Project.objects.create(name="Analysis")
  run = project1.add_pipeline(self.docker_pipeline_location)

scanpipe/tests/test_tasks.py

@@ -29,19 +29,19 @@
 
 
 class ScanPipeTasksTest(TestCase):
- pipeline_location = "scanpipe/pipelines/docker.py"
+ pipeline_location = "scanpipe/tests/pipelines/do_nothing.py"
 
  @mock.patch("scanpipe.pipelines.Pipeline.execute")
  def test_scanpipe_tasks_run_pipeline_task(self, mock_execute):
  project = Project.objects.create(name="my_project")
  run = project.add_pipeline(self.pipeline_location)
 
- mock_execute.return_value = (0, "mocked_output")
+ mock_execute.return_value = 0, ""
  tasks.run_pipeline_task(run.pk)
  mock_execute.assert_called_once()
 
  run.refresh_from_db()
  self.assertEqual(0, run.task_exitcode)
- self.assertEqual("mocked_output", run.task_output)
+ self.assertEqual("", run.task_output)
  self.assertIsNotNone(run.task_start_date)
  self.assertIsNotNone(run.task_end_date)