Add tests

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/tests/pipelines/v2_improvers/test_compute_advisory_content_hash.py

@@ -0,0 +1,88 @@
+#
+# 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
+
+import pytest
+
+from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.pipelines.v2_improvers.compute_advisory_content_hash import (
+    ComputeAdvisoryContentHash,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def advisory_factory():
+    def _create(count, with_hash=False, start=0):
+        objs = []
+        for i in range(start, start + count):
+            objs.append(
+                AdvisoryV2(
+                    summary=f"summary {i}",
+                    advisory_content_hash="existing_hash" if with_hash else None,
+                    unique_content_id=f"unique_id_{i}",
+                    advisory_id=f"ADV-{i}",
+                    datasource_id="ds",
+                    avid=f"ds/ADV-{i}",
+                    url=f"https://example.com/ADV-{i}",
+                )
+            )
+        return AdvisoryV2.objects.bulk_create(objs)
+
+    return _create
+
+
+def run_pipeline():
+    pipeline = ComputeAdvisoryContentHash()
+    pipeline.compute_advisory_content_hash()
+
+
+@patch(
+    "vulnerabilities.pipelines.v2_improvers.compute_advisory_content_hash.compute_advisory_content"
+)
+def test_pipeline_updates_only_missing_hash(mock_compute, advisory_factory):
+    advisory_factory(3, with_hash=False, start=0)
+    advisory_factory(2, with_hash=True, start=100)
+
+    mock_compute.return_value = "new_hash"
+
+    run_pipeline()
+
+    updated = AdvisoryV2.objects.filter(advisory_content_hash="new_hash").count()
+    untouched = AdvisoryV2.objects.filter(advisory_content_hash="existing_hash").count()
+
+    assert updated == 3
+    assert untouched == 2
+    assert mock_compute.call_count == 3
+
+
+@patch(
+    "vulnerabilities.pipelines.v2_improvers.compute_advisory_content_hash.compute_advisory_content"
+)
+def test_pipeline_bulk_update_batches(mock_compute, advisory_factory):
+    advisory_factory(6000, with_hash=False)
+
+    mock_compute.return_value = "batch_hash"
+
+    run_pipeline()
+
+    assert AdvisoryV2.objects.filter(advisory_content_hash="batch_hash").count() == 6000
+
+    assert mock_compute.call_count == 6000
+
+
+@patch(
+    "vulnerabilities.pipelines.v2_improvers.compute_advisory_content_hash.compute_advisory_content"
+)
+def test_pipeline_no_advisories(mock_compute):
+    run_pipeline()
+
+    assert mock_compute.call_count == 0

vulnerabilities/tests/test_api_v2.py

@@ -859,7 +859,7 @@ def setUp(self):
 
     def test_list_with_purl_filter(self):
         url = reverse("package-v3-list")
-        with self.assertNumQueries(29):
+        with self.assertNumQueries(31):
             response = self.client.get(url, {"purl": "pkg:pypi/sample@1.0.0"})
         assert response.status_code == 200
         assert "packages" in response.data["results"]
@@ -868,7 +868,7 @@ def test_list_with_purl_filter(self):
 
     def test_bulk_lookup(self):
         url = reverse("package-v3-bulk-lookup")
-        with self.assertNumQueries(28):
+        with self.assertNumQueries(30):
             response = self.client.post(url, {"purls": ["pkg:pypi/sample@1.0.0"]}, format="json")
         assert response.status_code == 200
         assert "packages" in response.data
@@ -878,7 +878,7 @@ def test_bulk_lookup(self):
     def test_bulk_search_plain(self):
         url = reverse("package-v3-bulk-search")
         payload = {"purls": ["pkg:pypi/sample@1.0.0"], "plain_purl": True, "purl_only": False}
-        with self.assertNumQueries(28):
+        with self.assertNumQueries(30):
             response = self.client.post(url, payload, format="json")
         assert response.status_code == 200
         assert "packages" in response.data
@@ -894,7 +894,7 @@ def test_bulk_search_purl_only(self):
 
     def test_lookup_single_package(self):
         url = reverse("package-v3-lookup")
-        with self.assertNumQueries(21):
+        with self.assertNumQueries(23):
             response = self.client.post(url, {"purl": "pkg:pypi/sample@1.0.0"}, format="json")
         assert response.status_code == 200
         assert any(pkg["purl"] == "pkg:pypi/sample@1.0.0" for pkg in response.data)