test: populate is_latest in old advisory fixture

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

Author: keshav-space

vulnerabilities/tests/pipelines/v2_improvers/test_collect_ssvc_trees.py

@@ -36,6 +36,7 @@ def vulnrichment_advisory(db):
         url="https://example.com/advisory/TEST-2024-0001",
         unique_content_id="unique-1234",
         date_collected=datetime.now(),
+        is_latest=True,
     )
 
 
@@ -59,6 +60,7 @@ def related_advisory(db):
         url="https://example.com/related/TEST-2024-0001",
         unique_content_id="unique-5678",
         date_collected=datetime.now(),
+        is_latest=True,
     )
 
 

vulnerabilities/tests/pipelines/v2_improvers/test_compute_package_risk_v2.py

@@ -34,6 +34,7 @@ def test_simple_risk_pipeline():
         unique_content_id="ajkef",
         url="https://test.com",
         date_collected=datetime.now(),
+        is_latest=True,
     )
     adv.save()
 

vulnerabilities/tests/pipelines/v2_improvers/test_relate_severities.py

@@ -25,6 +25,7 @@ def test_relate_severities_by_advisory_id():
         unique_content_id="ab1",
         url="https://example.com/advisory/CVE-2024-0001",
         date_collected="2024-01-01",
+        is_latest=True,
     )
 
     severity_advisory = AdvisoryV2.objects.create(
@@ -34,6 +35,7 @@ def test_relate_severities_by_advisory_id():
         unique_content_id="ab2",
         url="https://example.com/epss/CVE-2024-0001",
         date_collected="2024-01-02",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -59,6 +61,7 @@ def test_relate_severities_via_alias():
         unique_content_id="ab3",
         url="https://example.com/advisory/CVE-2024-0002",
         date_collected="2024-01-01",
+        is_latest=True,
     )
 
     base.aliases.create(alias="CVE-2024-ALIAS")
@@ -70,6 +73,7 @@ def test_relate_severities_via_alias():
         unique_content_id="ab4",
         url="https://example.com/epss/CVE-2024-ALIAS",
         date_collected="2024-01-02",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -91,6 +95,7 @@ def test_no_self_relation_created():
         url="https://example.com/advisory/CVE-2024-0003",
         date_collected="2024-01-03",
         avid="epss/CVE-2024-0003",
+        is_latest=True,
     )
     advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -112,6 +117,7 @@ def test_unsupported_severity_system_is_ignored():
         url="https://example.com/advisory/CVE-2024-0004",
         date_collected="2024-01-01",
         avid="nvd/CVE-2024-0004",
+        is_latest=True,
     )
 
     severity_advisory = AdvisoryV2.objects.create(
@@ -121,6 +127,7 @@ def test_unsupported_severity_system_is_ignored():
         url="https://example.com/epss/CVE-2024-0004",
         date_collected="2024-01-02",
         avid="epss/CVE-2024-0004",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system="UNKNOWN_SYSTEM",
@@ -142,6 +149,7 @@ def test_pipeline_is_idempotent():
         url="https://example.com/advisory/CVE-2024-0005",
         date_collected="2024-01-01",
         avid="nvd/CVE-2024-0005",
+        is_latest=True,
     )
 
     severity = AdvisoryV2.objects.create(
@@ -150,6 +158,7 @@ def test_pipeline_is_idempotent():
         unique_content_id="ab9",
         url="https://example.com/epss/CVE-2024-0005",
         date_collected="2024-01-02",
+        is_latest=True,
         avid="epss/CVE-2024-0005",
     )
     severity.severities.create(

vulnerabilities/tests/test_api_v3.py

@@ -14,22 +14,26 @@
 from rest_framework.test import APITestCase
 from univers.version_range import PypiVersionRange
 
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import PackageV2
 from vulnerabilities.pipes.advisory import insert_advisory_v2
+from vulnerabilities.tests.pipelines import TestLogger
 
 
 class APIV3TestCase(APITestCase):
     def setUp(self):
         from vulnerabilities.models import ImpactedPackage
 
-        self.advisory = AdvisoryV2.objects.create(
-            datasource_id="ghsa",
-            advisory_id="GHSA-1234",
-            avid="ghsa/GHSA-1234",
-            unique_content_id="f" * 64,
-            url="https://example.com/advisory",
-            date_collected="2025-07-01T00:00:00Z",
+        self.logger = TestLogger()
+        self.advisory = insert_advisory_v2(
+            advisory=AdvisoryDataV2(
+                summary="summary",
+                advisory_id="GHSA-1234",
+                url="https://example.com/advisory",
+            ),
+            pipeline_id="ghsa",
+            logger=self.logger.write,
         )
 
         self.package = PackageV2.objects.from_purl(purl="pkg:pypi/sample@1.0.0")

vulnerabilities/tests/test_same_avid_different_content_id.py

@@ -7,13 +7,14 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import uuid
-from datetime import timedelta
-
 import pytest
-from django.utils.timezone import now
 
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.pipes.advisory import insert_advisory_v2
+from vulnerabilities.tests.pipelines import TestLogger
+
+logger = TestLogger()
 
 
 @pytest.fixture
@@ -22,45 +23,29 @@ def advisory_factory(db):
     Factory to create AdvisoryV2 objects with minimal required fields.
     """
 
-    def _create(*, avid, advisory_id, collected_at):
-        return AdvisoryV2.objects.create(
-            datasource_id="test_source",
-            advisory_id=advisory_id,
-            avid=avid,
-            unique_content_id=str(uuid.uuid4()),
-            url="https://example.com/advisory",
-            date_collected=collected_at,
+    def _create(*, advisory_id, summary):
+
+        return insert_advisory_v2(
+            advisory=AdvisoryDataV2(
+                summary=summary,
+                advisory_id=advisory_id,
+                url="https://example.com/advisory",
+            ),
+            pipeline_id="source",
+            logger=logger.write,
         )
 
     return _create
 
 
-@pytest.fixture
-def timestamps():
-    now_ts = now()
-    return {
-        "old": now_ts - timedelta(days=3),
-        "mid": now_ts - timedelta(days=1),
-        "new": now_ts,
-    }
-
-
 @pytest.mark.django_db
 def test_latest_for_avid_returns_latest_by_date_collected(
-    advisory_factory, timestamps, django_assert_num_queries
+    advisory_factory, django_assert_num_queries
 ):
     avid = "source/ADV-1"
 
-    older = advisory_factory(
-        avid=avid,
-        advisory_id="ADV-1",
-        collected_at=timestamps["old"],
-    )
-    newer = advisory_factory(
-        avid=avid,
-        advisory_id="ADV-1",
-        collected_at=timestamps["new"],
-    )
+    older = advisory_factory(advisory_id="ADV-1", summary="old advisory")
+    newer = advisory_factory(advisory_id="ADV-1", summary="new advisory")
 
     with django_assert_num_queries(1):
         result = AdvisoryV2.objects.latest_for_avid(avid)
@@ -70,20 +55,11 @@ def test_latest_for_avid_returns_latest_by_date_collected(
 
 
 @pytest.mark.django_db
-def test_latest_for_avid_tie_breaks_by_id(advisory_factory, timestamps, django_assert_num_queries):
+def test_latest_for_avid_tie_breaks_by_id(advisory_factory, django_assert_num_queries):
     avid = "source/ADV-2"
-    ts = timestamps["mid"]
-
-    first = advisory_factory(
-        avid=avid,
-        advisory_id="ADV-2",
-        collected_at=ts,
-    )
-    second = advisory_factory(
-        avid=avid,
-        advisory_id="ADV-2",
-        collected_at=ts,
-    )
+
+    first = advisory_factory(advisory_id="ADV-2", summary="old advisory")
+    second = advisory_factory(advisory_id="ADV-2", summary="new advisory")
 
     with django_assert_num_queries(1):
         result = AdvisoryV2.objects.latest_for_avid(avid)
@@ -92,25 +68,11 @@ def test_latest_for_avid_tie_breaks_by_id(advisory_factory, timestamps, django_a
 
 
 @pytest.mark.django_db
-def test_latest_per_avid_returns_one_row_per_avid(
-    advisory_factory, timestamps, django_assert_num_queries
-):
-    advisory_factory(
-        avid="source/A",
-        advisory_id="A",
-        collected_at=timestamps["old"],
-    )
-    latest_a = advisory_factory(
-        avid="source/A",
-        advisory_id="A",
-        collected_at=timestamps["new"],
-    )
-
-    latest_b = advisory_factory(
-        avid="source/B",
-        advisory_id="B",
-        collected_at=timestamps["mid"],
-    )
+def test_latest_per_avid_returns_one_row_per_avid(advisory_factory, django_assert_num_queries):
+    advisory_factory(advisory_id="A", summary="old advisory")
+    latest_a = advisory_factory(advisory_id="A", summary="new advisory")
+
+    latest_b = advisory_factory(advisory_id="B", summary="new advisory")
 
     with django_assert_num_queries(1):
         qs = AdvisoryV2.objects.latest_per_avid()
@@ -122,19 +84,11 @@ def test_latest_per_avid_returns_one_row_per_avid(
 
 
 @pytest.mark.django_db
-def test_latest_per_avid_excludes_older_versions(advisory_factory, timestamps):
+def test_latest_per_avid_excludes_older_versions(advisory_factory):
     avid = "source/C"
 
-    older = advisory_factory(
-        avid=avid,
-        advisory_id="C",
-        collected_at=timestamps["old"],
-    )
-    latest = advisory_factory(
-        avid=avid,
-        advisory_id="C",
-        collected_at=timestamps["new"],
-    )
+    older = advisory_factory(advisory_id="C", summary="old advisory")
+    latest = advisory_factory(advisory_id="C", summary="new advisory")
 
     results = list(AdvisoryV2.objects.latest_per_avid())
 
@@ -144,30 +98,14 @@ def test_latest_per_avid_excludes_older_versions(advisory_factory, timestamps):
 
 @pytest.mark.django_db
 def test_latest_for_avids_filters_and_collapses_correctly(
-    advisory_factory, timestamps, django_assert_num_queries
+    advisory_factory, django_assert_num_queries
 ):
-    advisory_factory(
-        avid="source/A",
-        advisory_id="A",
-        collected_at=timestamps["old"],
-    )
-    latest_a = advisory_factory(
-        avid="source/A",
-        advisory_id="A",
-        collected_at=timestamps["new"],
-    )
-
-    latest_b = advisory_factory(
-        avid="source/B",
-        advisory_id="B",
-        collected_at=timestamps["mid"],
-    )
-
-    advisory_factory(
-        avid="source/C",
-        advisory_id="C",
-        collected_at=timestamps["new"],
-    )
+
+    advisory_factory(advisory_id="A", summary="old advisory")
+    latest_a = advisory_factory(advisory_id="A", summary="new advisory")
+
+    advisory_factory(advisory_id="B", summary="old advisory")
+    latest_b = advisory_factory(advisory_id="B", summary="new advisory")
 
     with django_assert_num_queries(1):
         qs = AdvisoryV2.objects.latest_for_avids({"source/A", "source/B"})