Add unit test for the get_initial_steps function #1351

Signed-off-by: tdruez <[email protected]>

aboutcode/pipeline/__init__.py

@@ -60,6 +60,9 @@ def get_steps(cls, groups=None):
 
         steps = cls.steps()
 
+        if initial_steps := cls.get_initial_steps():
+            steps = (*initial_steps, *steps)
+
         if groups is not None:
             steps = tuple(
                 step
@@ -175,13 +178,7 @@ def execute(self):
         """Execute each steps in the order defined on this pipeline class."""
         self.log(f"Pipeline [{self.pipeline_name}] starting")
 
-        # TODO: This is located on the PipelineDefinition class
         steps = self.pipeline_class.get_steps(groups=self.selected_groups)
-
-        # TODO: This is located on the PipelineDefinition class
-        if initial_steps := self.get_initial_steps():
-            steps = initial_steps + steps
-
         steps_count = len(steps)
         pipeline_start_time = timer()
 

scanpipe/tests/__init__.py

@@ -30,12 +30,14 @@
 from scanpipe.models import DiscoveredDependency
 from scanpipe.models import DiscoveredPackage
 from scanpipe.tests.pipelines.do_nothing import DoNothing
+from scanpipe.tests.pipelines.download_inputs import DownloadInput
 from scanpipe.tests.pipelines.profile_step import ProfileStep
 from scanpipe.tests.pipelines.raise_exception import RaiseException
 
 scanpipe_app = apps.get_app_config("scanpipe")
 
 scanpipe_app.register_pipeline("do_nothing", DoNothing)
+scanpipe_app.register_pipeline("download_inputs", DownloadInput)
 scanpipe_app.register_pipeline("profile_step", ProfileStep)
 scanpipe_app.register_pipeline("raise_exception", RaiseException)
 

scanpipe/tests/pipelines/do_nothing.py

@@ -30,6 +30,8 @@ class DoNothing(Pipeline):
     Description section of the doc string.
     """
 
+    download_inputs = False
+
     @classmethod
     def steps(cls):
         return (

scanpipe/tests/pipelines/download_inputs.py

@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/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://github.com/nexB/scancode.io for support and download.
+
+from scanpipe.pipelines import Pipeline
+
+
+class DownloadInput(Pipeline):
+    """Define the download_inputs attribute as True"""
+
+    download_inputs = True
+
+    @classmethod
+    def steps(cls):
+        return (cls.step1,)
+
+    def step1(self):
+        """Step1 doc."""
+        pass

scanpipe/tests/pipelines/raise_exception.py

@@ -26,6 +26,8 @@
 class RaiseException(Pipeline):
     """Raise an Exception."""
 
+    download_inputs = False
+
     @classmethod
     def steps(cls):
         return (cls.raise_exception_step,)

scanpipe/tests/pipelines/register_from_file.py

@@ -26,6 +26,8 @@
 class RegisterFromFile(DoNothing):
     """Register from its file path."""
 
+    download_inputs = False
+
     @classmethod
     def steps(cls):
         return (cls.step1,)

scanpipe/tests/pipelines/steps_as_attribute.py

@@ -26,6 +26,8 @@
 class StepsAsAttribute(Pipeline):
     """Declare steps as attribute."""
 
+    download_inputs = False
+
     def step1(self):
         return
 

scanpipe/tests/pipelines/with_groups.py

@@ -27,6 +27,8 @@
 class WithGroups(Pipeline):
     """Include "grouped" steps."""
 
+    download_inputs = False
+
     @classmethod
     def steps(cls):
         return (

scanpipe/tests/test_pipelines.py

@@ -39,6 +39,7 @@
 from scanpipe.models import CodebaseResource
 from scanpipe.models import DiscoveredPackage
 from scanpipe.models import Project
+from scanpipe.pipelines import CommonStepsMixin
 from scanpipe.pipelines import InputFilesError
 from scanpipe.pipelines import Pipeline
 from scanpipe.pipelines import deploy_to_develop
@@ -52,6 +53,7 @@
 from scanpipe.tests import make_package
 from scanpipe.tests import package_data1
 from scanpipe.tests.pipelines.do_nothing import DoNothing
+from scanpipe.tests.pipelines.download_inputs import DownloadInput
 from scanpipe.tests.pipelines.profile_step import ProfileStep
 from scanpipe.tests.pipelines.steps_as_attribute import StepsAsAttribute
 from scanpipe.tests.pipelines.with_groups import WithGroups
@@ -171,9 +173,13 @@ def test_scanpipe_pipeline_class_execute_with_selected_steps(self, step2, step1)
 
     def test_scanpipe_pipeline_class_download_inputs_attribute(self):
         project1 = Project.objects.create(name="Analysis")
-        run = project1.add_pipeline("do_nothing")
+        run = project1.add_pipeline("download_inputs")
         pipeline = run.make_pipeline_instance()
         self.assertTrue(pipeline.download_inputs)
+        expected = (CommonStepsMixin.download_missing_inputs,)
+        self.assertEqual(expected, pipeline.get_initial_steps())
+        expected = (CommonStepsMixin.download_missing_inputs, DownloadInput.step1)
+        self.assertEqual(expected, pipeline.get_steps())
         pipeline.execute()
         self.assertIn("Step [download_missing_inputs]", run.log)