test_populate_vulnerability_summary_pipeline.py

#
# 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.
#

import datetime
from pathlib import Path

import pytz
from django.test import TestCase

from vulnerabilities.models import Advisory
from vulnerabilities.models import Alias
from vulnerabilities.models import Vulnerability
from vulnerabilities.pipelines.populate_vulnerability_summary_pipeline import (
    PopulateVulnerabilitySummariesPipeline,
)


class PopulateVulnerabilitySummariesPipelineTest(TestCase):
    def setUp(self):
        self.data = Path(__file__).parent.parent / "test_data"

    def test_populate_missing_summaries_from_nvd(self):
        """
        Test that vulnerabilities without summaries get them from NVD advisories.
        """

        # Create a vulnerability without a summary
        vulnerability = Vulnerability.objects.create(
            vulnerability_id="VCID-1234",
            summary="",
        )
        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)

        # Create an NVD advisory with a summary
        adv = Advisory.objects.create(
            summary="Test vulnerability summary",
            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)

        # Run the pipeline
        pipeline = PopulateVulnerabilitySummariesPipeline()
        pipeline.populate_missing_summaries()

        # Check that the vulnerability now has a summary
        vulnerability.refresh_from_db()
        self.assertEqual(vulnerability.summary, "Test vulnerability summary")

    def test_no_matching_advisory(self):
        """
        Test handling of vulnerabilities that have no matching NVD advisory.
        """
        # Create a vulnerability without a summary
        vulnerability = Vulnerability.objects.create(
            vulnerability_id="VCID-1234",
            summary="",
        )
        Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)

        # Run the pipeline
        pipeline = PopulateVulnerabilitySummariesPipeline()
        pipeline.populate_missing_summaries()

        # Check that the vulnerability still has no summary
        vulnerability.refresh_from_db()
        self.assertEqual(vulnerability.summary, "")

    def test_vulnerability_without_alias(self):
        """
        Test handling of vulnerabilities that have no aliases.
        """

        # Create a vulnerability without a summary or alias
        vulnerability = Vulnerability.objects.create(
            vulnerability_id="VCID-1234",
            summary="",
        )

        # Run the pipeline
        pipeline = PopulateVulnerabilitySummariesPipeline()
        pipeline.populate_missing_summaries()

        # Check that the vulnerability still has no summary
        vulnerability.refresh_from_db()
        self.assertEqual(vulnerability.summary, "")

    def test_non_nvd_advisory_ignored(self):
        """
        Test that advisories from sources other than NVD are ignored.
        """

        # Create a vulnerability without a summary
        vulnerability = Vulnerability.objects.create(
            vulnerability_id="VCID-1234",
            summary="",
        )
        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)

        # Create a non-NVD advisory with a summary
        adv = Advisory.objects.create(
            summary="Test vulnerability summary",
            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)

        # Run the pipeline
        pipeline = PopulateVulnerabilitySummariesPipeline()
        pipeline.populate_missing_summaries()

        # Check that the vulnerability still has no summary
        vulnerability.refresh_from_db()
        self.assertEqual(vulnerability.summary, "")

    def test_multiple_matching_advisories(self):
        """
        Test that the most recent matching advisory is used when there are multiple.
        """
        vulnerability = Vulnerability.objects.create(
            vulnerability_id="VCID-1234",
            summary="",
        )
        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)

        # Create two NVD advisories with the same alias
        adv1 = Advisory.objects.create(
            summary="First matching advisory",
            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)

        adv2 = Advisory.objects.create(
            summary="Second matching advisory",
            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)

        # Run the pipeline
        pipeline = PopulateVulnerabilitySummariesPipeline()
        pipeline.populate_missing_summaries()

        # Check that the vulnerability now has the most recent summary
        vulnerability.refresh_from_db()
        self.assertEqual(vulnerability.summary, "Second matching advisory")