Update tests to support latest advisory model changes

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

Author: keshav-space

vulnerabilities/tests/pipelines/test_base_pipeline.py

@@ -39,16 +39,12 @@
 
 
 def get_advisory1(created_by="test_pipeline"):
-    adv = models.Advisory.objects.create(
-        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=created_by,
-        date_collected=timezone.now(),
+    from vulnerabilities.pipes.advisory import insert_advisory
+
+    return insert_advisory(
+        advisory=advisory_data1,
+        pipeline_id=created_by,
     )
-    adv.aliases.add(*get_or_create_aliases(advisory_data1.aliases))
-    return adv
 
 
 class TestVulnerableCodePipeline(TestCase):

vulnerabilities/tests/pipelines/test_remove_duplicate_advisories.py

@@ -39,6 +39,8 @@ def test_remove_duplicates_keeps_oldest(self):
         Test that when multiple advisories have the same content,
         only the oldest one is kept.
         """
+        from vulnerabilities.utils import compute_content_id
+
         # Create three advisories with same content but different dates
         dates = [
             datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
@@ -49,14 +51,15 @@ def test_remove_duplicates_keeps_oldest(self):
         advisories = []
         for date in dates:
             advisory = Advisory.objects.create(
+                unique_content_id=compute_content_id(advisory_data=self.advisory_data),
                 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],
                 date_imported=date,
                 date_collected=date,
+                created_by="test_pipeline",
             )
             advisories.append(advisory)
-            print(advisory.id)
 
         # Run the pipeline
         pipeline = RemoveDuplicateAdvisoriesPipeline()
@@ -73,19 +76,23 @@ def test_different_content_preserved(self):
         """
         # Create two advisories with different content
         advisory1 = Advisory.objects.create(
+            unique_content_id="test-id1",
             summary="Summary 1",
             affected_packages=[],
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
             references=[],
             date_imported=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
+            created_by="test_pipeline",
         )
 
         advisory2 = Advisory.objects.create(
+            unique_content_id="test-id2",
             summary="Summary 2",
             affected_packages=[],
             references=[],
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
             date_imported=datetime.datetime(2024, 1, 2, tzinfo=pytz.UTC),
+            created_by="test_pipeline",
         )
 
         # Run the pipeline
@@ -99,13 +106,16 @@ def test_recompute_content_ids(self):
         """
         Test that advisories without content IDs get them updated.
         """
+        from vulnerabilities.utils import compute_content_id
+
         # Create advisory without content ID
         advisory = Advisory.objects.create(
+            unique_content_id="incorrect-content-id",
             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],
-            unique_content_id="",
             date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
+            created_by="test_pipeline",
         )
 
         # Run the pipeline
@@ -114,4 +124,5 @@ def test_recompute_content_ids(self):
 
         # Check that content ID was updated
         advisory.refresh_from_db()
-        self.assertNotEqual(advisory.unique_content_id, "")
+        expected_content_id = compute_content_id(advisory_data=self.advisory_data)
+        self.assertNotEqual(advisory.unique_content_id, expected_content_id)

vulnerabilities/tests/pipes/test_advisory.py

@@ -18,6 +18,7 @@
 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",
@@ -34,16 +35,12 @@
 
 
 def get_advisory1(created_by="test_pipeline"):
-    advisory = models.Advisory.objects.create(
-        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=created_by,
-        date_collected=timezone.now(),
+    from vulnerabilities.pipes.advisory import insert_advisory
+
+    return insert_advisory(
+        advisory=advisory_data1,
+        pipeline_id=created_by,
     )
-    advisory.aliases.add(*get_or_create_aliases(advisory_data1.aliases))
-    return advisory
 
 
 def get_all_vulnerability_relationships_objects():

vulnerabilities/tests/test_add_cvsssv31.py

@@ -24,6 +24,7 @@ def setUp(self):
         self.pipeline = CVEAdvisoryMappingPipeline()
         advisory = Advisory.objects.create(
             created_by="nvd_importer",
+            unique_content_id="test-unique-content-id",
             references=[
                 {
                     "severities": [
@@ -41,6 +42,7 @@ def setUp(self):
             date_collected="2024-09-27T19:38:00Z",
         )
         advisory.aliases.add(*get_or_create_aliases(["CVE-2024-1234"]))
+
         vuln = Vulnerability.objects.create(vulnerability_id="CVE-2024-1234")
         sev = VulnerabilitySeverity.objects.create(
             scoring_system=CVSSV3.identifier,

vulnerabilities/tests/test_changelog.py

@@ -25,6 +25,7 @@ def test_package_changelog():
     pkg, _ = models.Package.objects.get_or_create_from_purl("pkg:npm/[email protected]")
     assert models.PackageChangeLog.objects.filter(package=pkg).count() == 0
     adv = models.Advisory.objects.create(
+        unique_content_id="test-unique-content-id1",
         created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST",
         date_collected=datetime.now(),
@@ -53,6 +54,7 @@ def test_package_changelog():
     pkg1, _ = models.Package.objects.get_or_create_from_purl("pkg:npm/[email protected]")
     assert models.PackageChangeLog.objects.filter(package=pkg1).count() == 0
     adv = models.Advisory.objects.create(
+        unique_content_id="test-unique-content-id2",
         created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST-1",
         date_collected=datetime.now(),
@@ -84,6 +86,7 @@ def test_package_changelog():
 @pytest.mark.django_db
 def test_vulnerability_changelog():
     adv = models.Advisory.objects.create(
+        unique_content_id="test-unique-content-id3",
         created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST_1",
         date_collected=datetime.now(),
@@ -117,6 +120,7 @@ def test_vulnerability_changelog():
 @pytest.mark.django_db
 def test_vulnerability_changelog_software_version():
     adv = models.Advisory.objects.create(
+        unique_content_id="test-unique-content-id4",
         created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST_1",
         date_collected=datetime.now(),

vulnerabilities/tests/test_compute_content_id.py

@@ -85,26 +85,6 @@ def test_different_metadata_same_content_same_id(self):
 
         assert compute_content_id(advisory1) == compute_content_id(advisory2)
 
-    def test_different_metadata_different_id_when_included(self):
-        """
-        Test that advisories with same content but different metadata have different content IDs
-        when include_metadata=True
-        """
-        advisory1 = self.base_advisory
-
-        advisory2 = AdvisoryData(
-            summary="Test summary",
-            affected_packages=self.base_advisory.affected_packages,
-            references=self.base_advisory.references,
-            date_published=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
-            url="https://different.url",
-        )
-
-        self.assertNotEqual(
-            compute_content_id(advisory1, include_metadata=True),
-            compute_content_id(advisory2, include_metadata=True),
-        )
-
     def test_different_summary_different_id(self):
         """
         Test that advisories with different summaries have different content IDs