Update tests related to V2 advisory

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

Author: keshav-space

vulnerabilities/tests/pipelines/test_collect_commits_v2.py

@@ -1,11 +1,19 @@
+#
+# 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 datetime import datetime
-from unittest.mock import patch
 
 import pytest
 
 from vulnerabilities.models import AdvisoryReference
 from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import CodeFixV2
+from vulnerabilities.models import ImpactedPackage
 from vulnerabilities.models import PackageV2
 from vulnerabilities.pipelines.v2_improvers.collect_commits import CollectFixCommitsPipeline
 from vulnerabilities.pipelines.v2_improvers.collect_commits import is_vcs_url
@@ -59,8 +67,8 @@ def test_is_vcs_url_already_processed_true():
         name="foo",
         version="1.0",
     )
-    advisory.affecting_packages.add(package)
-    advisory.save()
+    impact = ImpactedPackage.objects.create(advisory=advisory)
+    impact.affecting_packages.add(package)
     CodeFixV2.objects.create(
         commits=["https://github.com/user/repo/commit/abc123"],
         advisory=advisory,
@@ -87,9 +95,9 @@ def test_collect_fix_commits_pipeline_creates_entry():
     reference = AdvisoryReference.objects.create(
         url="https://github.com/test/testpkg/commit/abc123"
     )
-    advisory.affecting_packages.add(package)
+    impact = ImpactedPackage.objects.create(advisory=advisory)
+    impact.affecting_packages.add(package)
     advisory.references.add(reference)
-    advisory.save()
 
     pipeline = CollectFixCommitsPipeline()
     pipeline.collect_and_store_fix_commits()
@@ -117,13 +125,11 @@ def test_collect_fix_commits_pipeline_skips_non_commit_urls():
         name="otherpkg",
         version="2.0",
     )
-
-    advisory.affecting_packages.add(package)
+    impact = ImpactedPackage.objects.create(advisory=advisory)
+    impact.affecting_packages.add(package)
 
     reference = AdvisoryReference.objects.create(url="https://github.com/test/testpkg/issues/12")
-
     advisory.references.add(reference)
-    advisory.save()
 
     pipeline = CollectFixCommitsPipeline()
     pipeline.collect_and_store_fix_commits()

vulnerabilities/tests/pipelines/test_curl_importer_v2.py

@@ -16,7 +16,7 @@
 from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.pipelines.v2_importers.curl_importer import CurlImporterPipeline
 from vulnerabilities.pipelines.v2_importers.curl_importer import get_cwe_from_curl_advisory
 from vulnerabilities.pipelines.v2_importers.curl_importer import parse_curl_advisory
@@ -74,9 +74,9 @@ def test_collect_advisories(mock_fetch, pipeline):
 
     # Affected package check
     pkg = advisory.affected_packages[0]
-    assert isinstance(pkg, AffectedPackage)
+    assert isinstance(pkg, AffectedPackageV2)
     assert pkg.package == PackageURL(type="generic", namespace="curl.se", name="curl")
-    assert pkg.fixed_version == SemverVersion("8.7.0")
+    assert str(pkg.fixed_version_range) == "vers:generic/8.7.0"
     assert "8.6.0" in str(pkg.affected_version_range)
 
     # References

vulnerabilities/tests/pipelines/test_gitlab_v2_importer.py

@@ -99,7 +99,7 @@ def test_collect_advisories(mock_gitlab_yaml, mock_vcs_response, mock_fetch_via_
     assert advisory.summary == "Example vulnerability\nExample description"
     assert advisory.references_v2[0].url == "https://example.com/advisory"
     assert advisory.affected_packages[0].package.name == "package-name"
-    assert advisory.affected_packages[0].fixed_version
+    assert str(advisory.affected_packages[0].fixed_version_range) == "vers:pypi/2.0.0"
     assert advisory.weaknesses[0] == 79
 
 

vulnerabilities/tests/pipelines/test_istio_importer_v2.py

@@ -12,14 +12,8 @@
 from textwrap import dedent
 
 import pytest
-from packageurl import PackageURL
-from univers.version_constraint import VersionConstraint
-from univers.version_range import GitHubVersionRange
-from univers.version_range import GolangVersionRange
-from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.pipelines.v2_importers.istio_importer import IstioImporterPipeline
 
@@ -77,24 +71,11 @@ def test_istio_advisory_parsing():
             url="https://istio.io/latest/news/security/ISTIO-SECURITY-2019-002/",
         )
 
-        expected_versions = [
-            VersionConstraint(version=SemverVersion("1.0"), comparator=">="),
-            VersionConstraint(version=SemverVersion("1.0.8"), comparator="<="),
-            VersionConstraint(version=SemverVersion("1.1"), comparator=">="),
-            VersionConstraint(version=SemverVersion("1.1.9"), comparator="<="),
-            VersionConstraint(version=SemverVersion("1.2"), comparator=">="),
-            VersionConstraint(version=SemverVersion("1.2.1"), comparator="<="),
-        ]
-
-        expected_packages = [
-            AffectedPackage(
-                package=PackageURL(type="golang", namespace="istio.io", name="istio"),
-                affected_version_range=GolangVersionRange(constraints=expected_versions),
-            ),
-            AffectedPackage(
-                package=PackageURL(type="github", namespace="istio", name="istio"),
-                affected_version_range=GitHubVersionRange(constraints=expected_versions),
-            ),
-        ]
-
-        assert advisory.affected_packages == expected_packages
+        assert (
+            str(advisory.affected_packages[0].affected_version_range)
+            == "vers:github/>=1.0.0|<=1.0.8|>=1.1.0|<=1.1.9|>=1.2.0|<=1.2.1"
+        )
+        assert (
+            str(advisory.affected_packages[1].affected_version_range)
+            == "vers:golang/>=1.0.0|<=1.0.8|>=1.1.0|<=1.1.9|>=1.2.0|<=1.2.1"
+        )

vulnerabilities/tests/pipelines/test_mozilla_importer_v2.py

@@ -14,7 +14,6 @@
 from vulnerabilities.pipelines.v2_importers.mozilla_importer import get_severity_from_impact
 from vulnerabilities.pipelines.v2_importers.mozilla_importer import mfsa_id_from_filename
 from vulnerabilities.pipelines.v2_importers.mozilla_importer import parse_affected_packages
-from vulnerabilities.pipelines.v2_importers.mozilla_importer import parse_md_advisory
 from vulnerabilities.pipelines.v2_importers.mozilla_importer import parse_yml_advisory
 
 
@@ -57,7 +56,7 @@ def test_parse_affected_packages_valid():
     result = list(parse_affected_packages(packages))
     assert len(result) == 2
     assert result[0].package.name == "firefox"
-    assert str(result[0].fixed_version) == "89.0.0"
+    assert str(result[0].fixed_version_range) == "vers:generic/89.0.0"
 
 
 def test_parse_affected_packages_invalid():

vulnerabilities/tests/pipelines/test_npm_importer_pipeline_v2.py

@@ -102,7 +102,7 @@ def test_to_advisory_data_full(tmp_path):
     pkg = adv.affected_packages[0]
     assert pkg.package == PackageURL(type="npm", name="mypkg")
     assert isinstance(pkg.affected_version_range, NpmVersionRange)
-    assert pkg.fixed_version == SemverVersion("1.2.4")
+    assert str(pkg.fixed_version_range) == "vers:npm/>=1.2.4"
     assert set(adv.aliases) == {"CVE-123", "CVE-124"}
 
 
@@ -121,8 +121,8 @@ def test_get_affected_package_special_and_standard():
         {"vulnerable_versions": "<=99.999.99999", "patched_versions": "<0.0.0"}, "pkg"
     )
     assert isinstance(pkg.affected_version_range, NpmVersionRange)
-    assert pkg.fixed_version is None
+    assert pkg.fixed_version_range is None
     data2 = {"vulnerable_versions": "<=2.0.0", "patched_versions": ">=2.0.1"}
     pkg2 = p.get_affected_package(data2, "pkg2")
     assert isinstance(pkg2.affected_version_range, NpmVersionRange)
-    assert pkg2.fixed_version == SemverVersion("2.0.1")
+    assert str(pkg2.fixed_version_range) == "vers:npm/>=2.0.1"

vulnerabilities/tests/pipelines/test_postgresql_v2_importer.py

@@ -87,20 +87,20 @@ def test_collect_advisories(mock_get, importer):
     assert "Description of the issue" in advisory.summary
     assert len(advisory.references_v2) > 0
     assert advisory.affected_packages[0].package.name == "postgresql"
-    assert str(advisory.affected_packages[0].fixed_version) == "10.2"
+    assert str(advisory.affected_packages[0].fixed_version_range) == "vers:generic/10.2.0"
     assert advisory.affected_packages[0].affected_version_range.contains(SemverVersion("10.0.0"))
     assert advisory.affected_packages[0].affected_version_range.contains(SemverVersion("10.1.0"))
 
 
 @patch("vulnerabilities.pipelines.v2_importers.postgresql_importer.requests.get")
-def test_collect_advisories_with_no_fixed_version(mock_get, importer):
+def test_collect_advisories_with_no_fixed_version_range(mock_get, importer):
     mock_get.return_value.content = HTML_NO_FIX_ADVISORY.encode("utf-8")
     advisories = list(importer.collect_advisories())
 
     assert len(advisories) == 1
     advisory = advisories[0]
     assert advisory.advisory_id == "CVE-2023-5678"
-    assert advisory.affected_packages[0].fixed_version is None
+    assert advisory.affected_packages[0].fixed_version_range is None
     assert advisory.affected_packages[0].affected_version_range.contains(SemverVersion("9.5"))
     assert advisory.affected_packages[0].affected_version_range.contains(SemverVersion("9.6"))
 

vulnerabilities/tests/pipelines/test_vulnerablecode_importer_v2_pipeline.py

@@ -10,15 +10,11 @@
 import logging
 from datetime import datetime
 from datetime import timedelta
-from unittest import mock
 
 import pytest
-from packageurl import PackageURL
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import UnMergeablePackageError
 from vulnerabilities.models import AdvisoryV2
-from vulnerabilities.models import PackageV2
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 
 
@@ -64,117 +60,3 @@ def test_collect_and_store_advisories(dummy_importer):
     assert len(dummy_importer.log_messages) >= 2
     assert "Successfully collected" in dummy_importer.log_messages[-1][1]
     assert AdvisoryV2.objects.count() == 1
-
-
-def test_get_advisory_packages_basic(dummy_importer):
-    purl = PackageURL("pypi", None, "dummy", "1.0.0")
-    affected_package = mock.Mock()
-    affected_package.package = purl
-    dummy_importer.unfurl_version_ranges = False
-
-    with mock.patch(
-        "vulnerabilities.improvers.default.get_exact_purls", return_value=([purl], [purl])
-    ):
-        with mock.patch.object(
-            PackageV2.objects, "get_or_create_from_purl", return_value=(mock.Mock(), True)
-        ) as mock_get:
-            dummy_importer.get_advisory_packages(
-                advisory_data=mock.Mock(affected_packages=[affected_package])
-            )
-            assert mock_get.call_count == 2  # one affected, one fixed
-
-
-def test_get_published_package_versions_filters(dummy_importer):
-    purl = PackageURL("pypi", None, "example", None)
-
-    dummy_versions = [
-        mock.Mock(value="1.0.0", release_date=datetime.now() - timedelta(days=5)),
-        mock.Mock(value="2.0.0", release_date=datetime.now() + timedelta(days=5)),  # future
-    ]
-
-    with mock.patch(
-        "vulnerabilities.pipelines.package_versions.versions", return_value=dummy_versions
-    ):
-        versions = dummy_importer.get_published_package_versions(purl, until=datetime.now())
-        assert "1.0.0" in versions
-        assert "2.0.0" not in versions
-
-
-def test_get_published_package_versions_failure_logs(dummy_importer):
-    purl = PackageURL("pypi", None, "example", None)
-    with mock.patch(
-        "vulnerabilities.pipelines.package_versions.versions", side_effect=Exception("fail")
-    ):
-        versions = dummy_importer.get_published_package_versions(purl)
-        assert versions == []
-        assert any("Failed to fetch versions" in msg for lvl, msg in dummy_importer.log_messages)
-
-
-def test_expand_version_range_to_purls(dummy_importer):
-    purls = list(
-        dummy_importer.expand_verion_range_to_purls("npm", "lodash", "lodash", ["1.0.0", "1.1.0"])
-    )
-    assert all(isinstance(p, PackageURL) for p in purls)
-    assert purls[0].name == "lodash"
-
-
-def test_resolve_package_versions(dummy_importer):
-    dummy_importer.ignorable_versions = []
-    dummy_importer.expand_verion_range_to_purls = lambda *args, **kwargs: [
-        PackageURL("npm", None, "a", "1.0.0")
-    ]
-
-    with mock.patch(
-        "vulnerabilities.pipelines.resolve_version_range", return_value=(["1.0.0"], ["1.1.0"])
-    ), mock.patch(
-        "vulnerabilities.pipelines.get_affected_packages_by_patched_package",
-        return_value={None: [PackageURL("npm", None, "a", "1.0.0")]},
-    ), mock.patch(
-        "vulnerabilities.pipelines.nearest_patched_package", return_value=[]
-    ):
-        aff, fix = dummy_importer.resolve_package_versions(
-            affected_version_range=">=1.0.0",
-            pkg_type="npm",
-            pkg_namespace=None,
-            pkg_name="a",
-            valid_versions=["1.0.0", "1.1.0"],
-        )
-        assert any(isinstance(p, PackageURL) for p in aff)
-
-
-def test_get_impacted_packages_mergeable(dummy_importer):
-    ap = mock.Mock()
-    ap.package = PackageURL("npm", None, "abc", None)
-    dummy_importer.get_published_package_versions = lambda package_url, until: ["1.0.0", "1.1.0"]
-    dummy_importer.resolve_package_versions = lambda **kwargs: (
-        [PackageURL("npm", None, "abc", "1.0.0")],
-        [PackageURL("npm", None, "abc", "1.1.0")],
-    )
-
-    with mock.patch(
-        "vulnerabilities.importer.AffectedPackage.merge",
-        return_value=(ap.package, [">=1.0.0"], ["1.1.0"]),
-    ):
-        aff, fix = dummy_importer.get_impacted_packages([ap], datetime.now())
-        assert len(aff) == 1 and aff[0].version == "1.0.0"
-        assert len(fix) == 1 and fix[0].version == "1.1.0"
-
-
-def test_get_impacted_packages_unmergeable(dummy_importer):
-    ap = mock.Mock()
-    ap.package = PackageURL("npm", None, "abc", None)
-    ap.affected_version_range = ">=1.0.0"
-    ap.fixed_version = None
-
-    dummy_importer.get_published_package_versions = lambda package_url, until: ["1.0.0", "1.1.0"]
-    dummy_importer.resolve_package_versions = lambda **kwargs: (
-        [PackageURL("npm", None, "abc", "1.0.0")],
-        [PackageURL("npm", None, "abc", "1.1.0")],
-    )
-
-    with mock.patch(
-        "vulnerabilities.importer.AffectedPackage.merge", side_effect=UnMergeablePackageError
-    ):
-        aff, fix = dummy_importer.get_impacted_packages([ap], datetime.utcnow())
-        assert len(aff) == 1
-        assert aff[0].version == "1.0.0"

vulnerabilities/tests/pipelines/test_compute_package_risk_v2.py renamed to vulnerabilities/tests/pipelines/v2_improvers/test_compute_package_risk_v2.py

@@ -14,6 +14,7 @@
 from vulnerabilities.models import AdvisorySeverity
 from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import AdvisoryWeakness
+from vulnerabilities.models import ImpactedPackage
 from vulnerabilities.models import PackageV2
 from vulnerabilities.pipelines.v2_improvers.compute_package_risk import ComputePackageRiskPipeline
 from vulnerabilities.severity_systems import CVSSV3
@@ -54,8 +55,8 @@ def test_simple_risk_pipeline():
     weaknesses = AdvisoryWeakness.objects.create(cwe_id=119)
     adv.weaknesses.add(weaknesses)
 
-    adv.affecting_packages.add(pkg)
-    adv.save()
+    impact = ImpactedPackage.objects.create(advisory=adv)
+    impact.affecting_packages.add(pkg)
 
     improver = ComputePackageRiskPipeline()
     improver.execute()