Test advisory with duplicate content_id

Signed-off-by: Keshav Priyadarshi <[email protected]>

Author: keshav-space

vulnerabilities/import_runner.py

@@ -123,7 +123,7 @@ def process_advisories(
                 obj.aliases.add(*aliases)
                 if not obj.date_imported:
                     advisories.append(obj)
-            except Advisory.MultipleObjectsReturned as mo:
+            except Advisory.MultipleObjectsReturned:
                 logger.error(
                     f"Multiple Advisories returned: unique_content_id: {content_id}, url: {data.url}, advisory: {advisory!r}"
                 )

vulnerabilities/tests/conftest.py

@@ -25,7 +25,6 @@ def no_rmtree(monkeypatch):
 # Step 2: Run test for importer only if it is activated (pytestmark = pytest.mark.skipif(...))
 # Step 3: Migrate all the tests
 collect_ignore = [
-    "test_models.py",
     "test_rust.py",
     "test_suse_backports.py",
     "test_suse.py",

vulnerabilities/tests/pipelines/test_populate_vulnerability_summary_pipeline.py

@@ -43,6 +43,7 @@ def test_populate_missing_summaries_from_nvd(self):
             created_by="nvd_importer",
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
             unique_content_id="Test",
+            url="https://test.com",
         )
         adv.aliases.add(alias)
 
@@ -110,6 +111,7 @@ def test_non_nvd_advisory_ignored(self):
             created_by="other_importer",
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
             unique_content_id="Test",
+            url="https://test.com",
         )
 
         adv.aliases.add(alias)
@@ -138,6 +140,7 @@ def test_multiple_matching_advisories(self):
             created_by="nvd_importer",
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
             unique_content_id="Test",
+            url="https://test.com",
         )
 
         adv1.aliases.add(alias)
@@ -147,6 +150,7 @@ def test_multiple_matching_advisories(self):
             created_by="nvd_importer",
             date_collected=datetime.datetime(2024, 1, 2, tzinfo=pytz.UTC),
             unique_content_id="Test-1",
+            url="https://test.com",
         )
 
         adv2.aliases.add(alias)

vulnerabilities/tests/pipelines/test_remove_duplicate_advisories.py

@@ -32,6 +32,7 @@ def setUp(self):
                 )
             ],
             references=[Reference(url="https://example.com/vuln1")],
+            url="https://test.url/",
         )
 
     def test_remove_duplicates_keeps_oldest(self):
@@ -49,9 +50,10 @@ def test_remove_duplicates_keeps_oldest(self):
         ]
 
         advisories = []
-        for date in dates:
+        for i, date in enumerate(dates):
             advisory = Advisory.objects.create(
-                unique_content_id=compute_content_id(advisory_data=self.advisory_data),
+                unique_content_id=f"incorrect-content-id{i}",
+                url=self.advisory_data.url,
                 summary=self.advisory_data.summary,
                 affected_packages=[pkg.to_dict() for pkg in self.advisory_data.affected_packages],
                 references=[ref.to_dict() for ref in self.advisory_data.references],
@@ -77,6 +79,7 @@ def test_different_content_preserved(self):
         # Create two advisories with different content
         advisory1 = Advisory.objects.create(
             unique_content_id="test-id1",
+            url="https://test.url/",
             summary="Summary 1",
             affected_packages=[],
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
@@ -87,6 +90,7 @@ def test_different_content_preserved(self):
 
         advisory2 = Advisory.objects.create(
             unique_content_id="test-id2",
+            url="https://test.url/",
             summary="Summary 2",
             affected_packages=[],
             references=[],
@@ -111,6 +115,7 @@ def test_recompute_content_ids(self):
         # Create advisory without content ID
         advisory = Advisory.objects.create(
             unique_content_id="incorrect-content-id",
+            url=self.advisory_data.url,
             summary=self.advisory_data.summary,
             affected_packages=[pkg.to_dict() for pkg in self.advisory_data.affected_packages],
             references=[ref.to_dict() for ref in self.advisory_data.references],
@@ -125,4 +130,4 @@ def test_recompute_content_ids(self):
         # Check that content ID was updated
         advisory.refresh_from_db()
         expected_content_id = compute_content_id(advisory_data=self.advisory_data)
-        self.assertNotEqual(advisory.unique_content_id, expected_content_id)
+        self.assertEqual(advisory.unique_content_id, expected_content_id)

vulnerabilities/tests/pipes/test_advisory.py

@@ -7,7 +7,10 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import pytest
+from datetime import datetime
+
+from django.core.exceptions import ValidationError
+from django.test import TestCase
 from django.utils import timezone
 from packageurl import PackageURL
 from univers.version_range import VersionRange
@@ -18,65 +21,116 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.pipes.advisory import get_or_create_aliases
 from vulnerabilities.pipes.advisory import import_advisory
+from vulnerabilities.utils import compute_content_id
 
-advisory_data1 = AdvisoryData(
-    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",
-)
 
+class TestPipeAdvisory(TestCase):
+    def setUp(self):
+        self.advisory_data1 = AdvisoryData(
+            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",
+        )
 
-def get_advisory1(created_by="test_pipeline"):
-    from vulnerabilities.pipes.advisory import insert_advisory
+    def get_advisory1(self, created_by="test_pipeline"):
+        from vulnerabilities.pipes.advisory import insert_advisory
 
-    return insert_advisory(
-        advisory=advisory_data1,
-        pipeline_id=created_by,
-    )
+        return insert_advisory(
+            advisory=self.advisory_data1,
+            pipeline_id=created_by,
+        )
 
+    def get_all_vulnerability_relationships_objects(self):
+        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()),
+        }
 
-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()),
-    }
+    def test_vulnerability_pipes_importer_import_advisory(self):
+        advisory1 = self.get_advisory1(created_by="test_importer_pipeline")
+        import_advisory(advisory=advisory1, pipeline_id="test_importer_pipeline")
+        all_vulnerability_relation_objects = self.get_all_vulnerability_relationships_objects()
+        import_advisory(advisory=advisory1, pipeline_id="test_importer_pipeline")
+        assert (
+            all_vulnerability_relation_objects == self.get_all_vulnerability_relationships_objects()
+        )
 
+    def test_vulnerability_pipes_importer_import_advisory_different_pipelines(self):
+        advisory1 = self.get_advisory1(created_by="test_importer_pipeline")
+        import_advisory(advisory=advisory1, pipeline_id="test_importer1_pipeline")
+        all_vulnerability_relation_objects = self.get_all_vulnerability_relationships_objects()
+        import_advisory(advisory=advisory1, pipeline_id="test_importer2_pipeline")
+        assert (
+            all_vulnerability_relation_objects == self.get_all_vulnerability_relationships_objects()
+        )
 
-@pytest.mark.django_db
-def test_vulnerability_pipes_importer_import_advisory():
-    advisory1 = get_advisory1(created_by="test_importer_pipeline")
-    import_advisory(advisory=advisory1, pipeline_id="test_importer_pipeline")
-    all_vulnerability_relation_objects = get_all_vulnerability_relationships_objects()
-    import_advisory(advisory=advisory1, pipeline_id="test_importer_pipeline")
-    assert all_vulnerability_relation_objects == get_all_vulnerability_relationships_objects()
+    def test_vulnerability_pipes_get_or_create_aliases(self):
+        aliases = ["CVE-TEST-123", "CVE-TEST-124"]
+        result_aliases_qs = get_or_create_aliases(aliases=aliases)
+        result_aliases = [i.alias for i in result_aliases_qs]
+        assert 2 == result_aliases_qs.count()
+        assert "CVE-TEST-123" in result_aliases
+        assert "CVE-TEST-124" in result_aliases
 
+    def test_advisory_insert_without_url(self):
+        with self.assertRaises(ValidationError):
+            date = datetime.now()
+            models.Advisory.objects.create(
+                unique_content_id=compute_content_id(advisory_data=self.advisory_data1),
+                summary=self.advisory_data1.summary,
+                affected_packages=[pkg.to_dict() for pkg in self.advisory_data1.affected_packages],
+                references=[ref.to_dict() for ref in self.advisory_data1.references],
+                date_imported=date,
+                date_collected=date,
+                created_by="test_pipeline",
+            )
 
-@pytest.mark.django_db
-def test_vulnerability_pipes_importer_import_advisory_different_pipelines():
-    advisory1 = get_advisory1(created_by="test_importer_pipeline")
-    import_advisory(advisory=advisory1, pipeline_id="test_importer1_pipeline")
-    all_vulnerability_relation_objects = get_all_vulnerability_relationships_objects()
-    import_advisory(advisory=advisory1, pipeline_id="test_importer2_pipeline")
-    assert all_vulnerability_relation_objects == get_all_vulnerability_relationships_objects()
+    def test_advisory_insert_without_content_id(self):
+        with self.assertRaises(ValidationError):
+            date = datetime.now()
+            models.Advisory.objects.create(
+                url=self.advisory_data1.url,
+                summary=self.advisory_data1.summary,
+                affected_packages=[pkg.to_dict() for pkg in self.advisory_data1.affected_packages],
+                references=[ref.to_dict() for ref in self.advisory_data1.references],
+                date_imported=date,
+                date_collected=date,
+                created_by="test_pipeline",
+            )
 
+    def test_advisory_insert_no_duplicate_content_id(self):
+        date = datetime.now()
+        models.Advisory.objects.create(
+            unique_content_id=compute_content_id(advisory_data=self.advisory_data1),
+            url=self.advisory_data1.url,
+            summary=self.advisory_data1.summary,
+            affected_packages=[pkg.to_dict() for pkg in self.advisory_data1.affected_packages],
+            references=[ref.to_dict() for ref in self.advisory_data1.references],
+            date_imported=date,
+            date_collected=date,
+            created_by="test_pipeline",
+        )
 
-@pytest.mark.django_db
-def test_vulnerability_pipes_get_or_create_aliases():
-    aliases = ["CVE-TEST-123", "CVE-TEST-124"]
-    result_aliases_qs = get_or_create_aliases(aliases=aliases)
-    result_aliases = [i.alias for i in result_aliases_qs]
-    assert 2 == result_aliases_qs.count()
-    assert "CVE-TEST-123" in result_aliases
-    assert "CVE-TEST-124" in result_aliases
+        with self.assertRaises(ValidationError):
+            models.Advisory.objects.create(
+                unique_content_id=compute_content_id(advisory_data=self.advisory_data1),
+                url=self.advisory_data1.url,
+                summary=self.advisory_data1.summary,
+                affected_packages=[pkg.to_dict() for pkg in self.advisory_data1.affected_packages],
+                references=[ref.to_dict() for ref in self.advisory_data1.references],
+                date_imported=date,
+                date_collected=date,
+                created_by="test_pipeline",
+            )

vulnerabilities/tests/test_add_cvsssv31.py

@@ -25,6 +25,7 @@ def setUp(self):
         advisory = Advisory.objects.create(
             created_by="nvd_importer",
             unique_content_id="test-unique-content-id",
+            url="https://nvd.nist.gov/vuln/detail/CVE-2024-1234",
             references=[
                 {
                     "severities": [

vulnerabilities/tests/test_import_runner.py

@@ -179,6 +179,7 @@ def test_advisory_summary_clean_up():
 
 DUMMY_ADVISORY = models.Advisory(
     unique_content_id="test-unique-content-id",
+    url="https://test.url/",
     summary="dummy",
     created_by="tests",
     date_collected=timezone.now(),