Fix the get_pipeline_class to only return subclass of Pipeline #11

docs/changelog.rst

@@ -1,6 +1,11 @@
 // Release notes
 // -------------
 
+### v1.0.2 (unreleased)
+
+- *2020-09-17* **ScanPipe** -- Always return the Pipeline subclass/implementation
+  from the module inspection. Fix for https://github.com/nexB/scancode.io/issues/11
+
 ### v1.0.1 (2020-09-12)
 
 - *2020-09-11* **ScanPipe** -- Do not fail when collecting system packages in

scanpipe/pipelines/__init__.py

@@ -88,8 +88,12 @@ def output_dot(self, direction="TB", simplify=False):
         return output
 
 
-def is_pipeline_class(obj):
-    return inspect.isclass(obj) and issubclass(obj, Pipeline)
+def is_pipeline_subclass(obj):
+    """
+    Return True if the `obj` is a subclass of `Pipeline` except for the
+    `Pipeline` class itself.
+    """
+    return inspect.isclass(obj) and issubclass(obj, Pipeline) and obj is not Pipeline
 
 
 def get_pipeline_class(pipeline_location):
@@ -98,7 +102,7 @@ def get_pipeline_class(pipeline_location):
     """
     module_name = pipeline_location.replace(".py", "").replace("/", ".")
     module = importlib.import_module(module_name)
-    module_classes = inspect.getmembers(module, is_pipeline_class)
+    module_classes = inspect.getmembers(module, is_pipeline_subclass)
     _, pipeline_class = [cls for cls in module_classes][0]
 
     return pipeline_class

scanpipe/tests/test_pipelines.py

@@ -28,27 +28,43 @@
 from scanpipe.pipelines import get_pipeline_class
 from scanpipe.pipelines import get_pipeline_description
 from scanpipe.pipelines import get_pipeline_doc
+from scanpipe.pipelines import is_pipeline_subclass
 from scanpipe.pipelines.docker import DockerPipeline
+from scanpipe.pipelines.root_filesystems import RootfsPipeline
+from scanpipe.pipelines.scan_inventory import CollectInventoryFromScanCodeScan
 
 
 class ScanPipeModelsTest(TestCase):
-    pipeline_location = "scanpipe/pipelines/docker.py"
+    docker_pipeline_location = "scanpipe/pipelines/docker.py"
+    rootfs_pipeline_location = "scanpipe/pipelines/root_filesystems.py"
+    scan_pipeline_location = "scanpipe/pipelines/scan_inventory.py"
 
     def test_scanpipe_pipeline_class_get_project_instance(self):
         project1 = Project.objects.create(name="Analysis")
         project_instance = Pipeline.get_project_instance(project_pk=project1.pk)
         self.assertEqual(project1, project_instance)
 
+    def test_scanpipe_pipelines_is_pipeline_subclass(self):
+        self.assertFalse(is_pipeline_subclass(None))
+        self.assertFalse(is_pipeline_subclass(Pipeline))
+        self.assertTrue(is_pipeline_subclass(DockerPipeline))
+        self.assertTrue(is_pipeline_subclass(RootfsPipeline))
+        self.assertTrue(is_pipeline_subclass(CollectInventoryFromScanCodeScan))
+
     def test_scanpipe_pipelines_get_pipeline_class(self):
-        pipeline_class = get_pipeline_class(self.pipeline_location)
+        pipeline_class = get_pipeline_class(self.docker_pipeline_location)
         self.assertEqual(DockerPipeline, pipeline_class)
+        pipeline_class = get_pipeline_class(self.rootfs_pipeline_location)
+        self.assertEqual(RootfsPipeline, pipeline_class)
+        pipeline_class = get_pipeline_class(self.scan_pipeline_location)
+        self.assertEqual(CollectInventoryFromScanCodeScan, pipeline_class)
 
     def test_scanpipe_pipelines_get_pipeline_doc(self):
-        doc = get_pipeline_doc(self.pipeline_location)
+        doc = get_pipeline_doc(self.docker_pipeline_location)
         self.assertEqual("A pipeline to analyze a Docker image.", doc)
 
     def test_scanpipe_pipelines_get_pipeline_description(self):
-        description = get_pipeline_description(self.pipeline_location)
+        description = get_pipeline_description(self.docker_pipeline_location)
         self.assertIn("executing DockerPipeline for user:", description)
         self.assertIn("A pipeline to analyze a Docker image.", description)
         self.assertIn("Step start", description)
@@ -57,7 +73,7 @@ def test_scanpipe_pipelines_get_pipeline_description(self):
         self.assertIn("Analysis completed.", description)
 
     def test_scanpipe_pipelines_pipeline_graph_output_dot(self):
-        pipeline_class = get_pipeline_class(self.pipeline_location)
+        pipeline_class = get_pipeline_class(self.docker_pipeline_location)
         pipeline_graph = PipelineGraph(pipeline_class)
         output_dot = pipeline_graph.output_dot()
         self.assertIn("rankdir=TB;", output_dot)