Add test for base and pypa importer pipeline

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/import_runner.py

@@ -18,7 +18,6 @@
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Importer
-from vulnerabilities.importers import IMPORTERS_REGISTRY
 from vulnerabilities.improver import Inference
 from vulnerabilities.improvers.default import DefaultImporter
 from vulnerabilities.models import Advisory

vulnerabilities/pipelines/__init__.py

@@ -3,9 +3,10 @@
 # VulnerableCode is a trademark of nexB Inc.
 # SPDX-License-Identifier: Apache-2.0
 # See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
-# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
 import logging
 from datetime import datetime
 from datetime import timezone
@@ -17,8 +18,8 @@
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.improver import MAX_CONFIDENCE
-from vulnerabilities.pipelines.pipes.importer import import_advisory
-from vulnerabilities.pipelines.pipes.importer import insert_advisory
+from vulnerabilities.pipes.importer import import_advisory
+from vulnerabilities.pipes.importer import insert_advisory
 from vulnerabilities.utils import classproperty
 
 module_logger = logging.getLogger(__name__)

…nerabilities/pipelines/pipes/importer.py vulnerabilities/pipes/importer.pyvulnerabilities/pipelines/pipes/importer.py renamed to vulnerabilities/pipes/importer.py vulnerabilities/pipelines/pipes/importer.py renamed to vulnerabilities/pipes/importer.py

@@ -3,9 +3,10 @@
 # VulnerableCode is a trademark of nexB Inc.
 # SPDX-License-Identifier: Apache-2.0
 # See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
-# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
 import logging
 from datetime import datetime
 from datetime import timezone
@@ -14,10 +15,8 @@
 
 from django.db import transaction
 
-from vulnerabilities import import_runner
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.improver import MAX_CONFIDENCE
-from vulnerabilities.improvers import default
 from vulnerabilities.models import Advisory
 from vulnerabilities.models import Package
 from vulnerabilities.models import PackageRelatedVulnerability
@@ -55,18 +54,21 @@ def insert_advisory(advisory: AdvisoryData, pipeline_name: str, logger: Callable
 def import_advisory(
     advisory: Advisory,
     pipeline_name: str,
-    logger: Callable,
     confidence: int = MAX_CONFIDENCE,
+    logger: Callable = None,
 ):
     """
     Create initial Vulnerability Package relationships for the advisory,
     including references and severity scores.
 
     Package relationships are established only for resolved (concrete) versions.
     """
+    from vulnerabilities import import_runner
+    from vulnerabilities.improvers import default
 
     advisory_data: AdvisoryData = advisory.to_advisory_data()
-    logger(f"Importing advisory id: {advisory.id}", level=logging.DEBUG)
+    if logger:
+        logger(f"Importing advisory id: {advisory.id}", level=logging.DEBUG)
 
     affected_purls = []
     fixed_purls = []
@@ -85,7 +87,8 @@ def import_advisory(
     )
 
     if not vulnerability:
-        logger(f"Unable to get vulnerability for advisory: {advisory!r}", level=logging.WARNING)
+        if logger:
+            logger(f"Unable to get vulnerability for advisory: {advisory!r}", level=logging.WARNING)
         return
 
     for ref in advisory_data.references:
@@ -118,16 +121,18 @@ def import_advisory(
                     },
                 )
             except:
-                logger(
-                    f"Failed to create VulnerabilitySeverity for: {severity} with error:\n{traceback_format_exc()}",
-                    level=logging.ERROR,
-                )
+                if logger:
+                    logger(
+                        f"Failed to create VulnerabilitySeverity for: {severity} with error:\n{traceback_format_exc()}",
+                        level=logging.ERROR,
+                    )
             if not created:
-                logger(
-                    f"Severity updated for reference {ref!r} to value: {severity.value!r} "
-                    f"and scoring_elements: {severity.scoring_elements!r}",
-                    level=logging.DEBUG,
-                )
+                if logger:
+                    logger(
+                        f"Severity updated for reference {ref!r} to value: {severity.value!r} "
+                        f"and scoring_elements: {severity.scoring_elements!r}",
+                        level=logging.DEBUG,
+                    )
 
     for affected_purl in affected_purls or []:
         vulnerable_package, _ = Package.objects.get_or_create_from_purl(purl=affected_purl)

vulnerabilities/tests/__init__.py

@@ -3,6 +3,52 @@
 # VulnerableCode is a trademark of nexB Inc.
 # SPDX-License-Identifier: Apache-2.0
 # See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
-# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
+from django.utils import timezone
+from packageurl import PackageURL
+from univers.version_range import VersionRange
+
+from vulnerabilities import models
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import Reference
+
+advisory_data1 = AdvisoryData(
+    aliases=["CVE-2020-13371337"],
+    summary="vulnerability description here",
+    affected_packages=[
+        AffectedPackage(
+            package=PackageURL(type="pypi", name="dummy"),
+            affected_version_range=VersionRange.from_string("vers:pypi/>=1.0.0|<=2.0.0"),
+        )
+    ],
+    references=[Reference(url="https://example.com/with/more/info/CVE-2020-13371337")],
+    date_published=timezone.now(),
+    url="https://test.com",
+)
+
+
+advisory1 = models.Advisory(
+    aliases=advisory_data1.aliases,
+    summary=advisory_data1.summary,
+    affected_packages=[pkg.to_dict() for pkg in advisory_data1.affected_packages],
+    references=[ref.to_dict() for ref in advisory_data1.references],
+    url=advisory_data1.url,
+    created_by="tests",
+    date_collected=timezone.now(),
+)
+
+
+def get_all_vulnerability_relationships_objects():
+    return {
+        "vulnerabilities": list(models.Vulnerability.objects.all()),
+        "aliases": list(models.Alias.objects.all()),
+        "references": list(models.VulnerabilityReference.objects.all()),
+        "advisories": list(models.Advisory.objects.all()),
+        "packages": list(models.Package.objects.all()),
+        "references": list(models.VulnerabilityReference.objects.all()),
+        "severity": list(models.VulnerabilitySeverity.objects.all()),
+    }

vulnerabilities/tests/pipelines/test_base_pipeline.py

@@ -0,0 +1,63 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from vulnerabilities import models
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
+from vulnerabilities.tests import advisory1
+from vulnerabilities.tests import advisory_data1
+
+
+class TestVulnerableCodeBaseImporterPipeline(TestCase):
+    @patch.object(
+        VulnerableCodeBaseImporterPipeline,
+        "collect_advisories",
+        return_value=[advisory_data1],
+    )
+    @patch.object(
+        VulnerableCodeBaseImporterPipeline,
+        "advisories_count",
+        return_value=1,
+    )
+    def test_collect_and_store_advisories(self, mock_advisories_count, mock_collect_advisories):
+        self.assertEqual(0, models.Advisory.objects.count())
+
+        base_pipeline = VulnerableCodeBaseImporterPipeline()
+        base_pipeline.collect_and_store_advisories()
+
+        mock_advisories_count.assert_called_once()
+        mock_collect_advisories.assert_called_once()
+
+        self.assertEqual(1, models.Advisory.objects.count())
+
+        collected_advisory = models.Advisory.objects.first()
+        result_aliases = collected_advisory.aliases
+        expected_aliases = advisory_data1.aliases
+
+        self.assertEqual(expected_aliases, result_aliases)
+        self.assertEqual(base_pipeline.qualified_name, collected_advisory.created_by)
+
+    def test_import_new_advisories(self):
+        self.assertEqual(0, models.Vulnerability.objects.count())
+
+        base_pipeline = VulnerableCodeBaseImporterPipeline()
+        base_pipeline.new_advisories = [advisory1]
+        base_pipeline.import_new_advisories()
+
+        self.assertEqual(1, models.Vulnerability.objects.count())
+
+        imported_vulnerability = models.Vulnerability.objects.first()
+
+        self.assertEqual(1, imported_vulnerability.aliases.count())
+
+        expected_alias = imported_vulnerability.aliases.first()
+        self.assertEqual(advisory1.aliases[0], expected_alias.alias)

vulnerabilities/tests/pipelines/test_pypa_importer_pipeline.py

@@ -3,25 +3,27 @@
 # VulnerableCode is a trademark of nexB Inc.
 # SPDX-License-Identifier: Apache-2.0
 # See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
-# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+
 import os
+from pathlib import Path
 from unittest import TestCase
 
 import saneyaml
 
 from vulnerabilities.importers.osv import parse_advisory_data
 from vulnerabilities.tests import util_tests
 
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TEST_DATA = os.path.join(BASE_DIR, "test_data/pypa")
+TEST_DATA = data = Path(__file__).parent.parent / "test_data" / "pypa"
 
 
-class TestPyPaImporter(TestCase):
+class TestPyPaImporterPipeline(TestCase):
     def test_to_advisories_with_summary(self):
-        with open(os.path.join(TEST_DATA, "pypa_test.yaml")) as f:
-            mock_response = saneyaml.load(f)
+        pypa_advisory_path = TEST_DATA / "pypa_test.yaml"
+
+        mock_response = saneyaml.load(pypa_advisory_path.read_text())
         expected_file = os.path.join(TEST_DATA, "pypa-expected.json")
         imported_data = parse_advisory_data(
             mock_response,

vulnerabilities/tests/pipes/test_importer.py

@@ -0,0 +1,30 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import pytest
+
+from vulnerabilities.pipes.importer import import_advisory
+from vulnerabilities.tests import advisory1
+from vulnerabilities.tests import get_all_vulnerability_relationships_objects
+
+
+@pytest.mark.django_db
+def test_vulnerability_pipes_importer_import_advisory():
+    import_advisory(advisory=advisory1, pipeline_name="test_importer_pipeline")
+    all_vulnerability_relation_objects = get_all_vulnerability_relationships_objects()
+    import_advisory(advisory=advisory1, pipeline_name="test_importer_pipeline")
+    assert all_vulnerability_relation_objects == get_all_vulnerability_relationships_objects()
+
+
+@pytest.mark.django_db
+def test_vulnerability_pipes_importer_import_advisory_different_pipelines():
+    import_advisory(advisory=advisory1, pipeline_name="test_importer1_pipeline")
+    all_vulnerability_relation_objects = get_all_vulnerability_relationships_objects()
+    import_advisory(advisory=advisory1, pipeline_name="test_importer2_pipeline")
+    assert all_vulnerability_relation_objects == get_all_vulnerability_relationships_objects()