test_package_list_parser.py

# Copyright (C) 2021 Intel Corporation
# SPDX-License-Identifier: GPL-3.0-or-later

import json
import subprocess
import unittest.mock as mock
from pathlib import Path

import distro
import pytest

from cve_bin_tool.error_handler import ErrorMode
from cve_bin_tool.package_list_parser import (
    SUPPORTED_DISTROS,
    EmptyTxtError,
    InvalidListError,
    PackageListParser,
    Remarks,
)
from cve_bin_tool.util import ProductInfo


class TestPackageListParser:
    """
    Tests for cve_bin_tool/package_list_parser.py
    It handles parsing of package data on specific linux distros.
    """

    TXT_PATH = Path(__file__).parent.resolve() / "txt"

    REQ_PARSED_TRIAGE_DATA = {
        ProductInfo(
            vendor="httplib2_project*",
            product="httplib2",
            version="0.18.1",
            location="/usr/local/bin/httplib2",
        ): {
            "default": {"remarks": Remarks.NewFound, "comments": "", "severity": ""},
            "paths": {""},
        },
        ProductInfo(
            vendor="python*",
            product="requests",
            version="2.25.1",
            location="/usr/local/bin/requests",
        ): {
            "default": {"remarks": Remarks.NewFound, "comments": "", "severity": ""},
            "paths": {""},
        },
    }

    # Find the versions of the ubuntu packages
    UBUNTU_PACKAGE_VERSIONS = (
        (
            subprocess.run(
                [
                    "dpkg-query",
                    "--show",
                    "--showformat=${Version}\n",
                    "bash",
                    "binutils",
                    "wget",
                ],
                stdout=subprocess.PIPE,
            )
            .stdout.decode("utf-8")
            .splitlines()
        )
        if "ubuntu" in distro.id()
        else ["dummy", "array", "for windows"]
    )

    UBUNTU_PARSED_TRIAGE_DATA = {
        ProductInfo(
            vendor="gnu*",
            product="bash",
            version=UBUNTU_PACKAGE_VERSIONS[0],
            location="/usr/local/bin/bash",
        ): {
            "default": {"remarks": Remarks.NewFound, "comments": "", "severity": ""},
            "paths": {""},
        },
        ProductInfo(
            vendor="gnu*",
            product="binutils",
            version=UBUNTU_PACKAGE_VERSIONS[1],
            location="/usr/local/bin/binutils",
        ): {
            "default": {"remarks": Remarks.NewFound, "comments": "", "severity": ""},
            "paths": {""},
        },
        ProductInfo(
            vendor="gnu*",
            product="wget",
            version=UBUNTU_PACKAGE_VERSIONS[2],
            location="/usr/local/bin/wget",
        ): {
            "default": {"remarks": Remarks.NewFound, "comments": "", "severity": ""},
            "paths": {""},
        },
    }

    @pytest.mark.parametrize("filepath", [str(TXT_PATH / "nonexistent.txt")])
    def test_nonexistent_txt(self, filepath):
        """Test behaviour on non-existent file"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        with pytest.raises(FileNotFoundError):
            package_list.parse_list()

    @pytest.mark.parametrize(
        "filepath, exception", [(str(TXT_PATH / "empty.txt"), EmptyTxtError)]
    )
    def test_empty_txt(self, filepath, exception):
        """Test an empty list"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        with pytest.raises(exception):
            package_list.parse_list()

    @pytest.mark.parametrize(
        "filepath, exception", [(str(TXT_PATH / "not_txt.csv"), InvalidListError)]
    )
    def test_not_txt(self, filepath, exception):
        """Test an invalid type of list"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        with pytest.raises(exception):
            package_list.parse_list()

    # @pytest.mark.skipif(
    #     "ubuntu" not in distro.id(),
    #     reason="Test for Ubuntu systems",
    # )
    @pytest.mark.skip(reason="Test is broken, needs fixing")
    @pytest.mark.parametrize(
        "filepath, parsed_data",
        [(str(TXT_PATH / "test_requirements.txt"), REQ_PARSED_TRIAGE_DATA)],
    )
    def test_valid_requirements(self, filepath, parsed_data):
        """Test a valid requirements list"""
        # packages is installed from test_requirements with specific versions for the test to pass
        subprocess.run(["pip", "install", "-r", filepath])
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        assert package_list.parse_list() == parsed_data
        # Update the packages back to latest
        subprocess.run(["pip", "install", "httplib2", "requests", "-U"])

    @pytest.mark.skipif(
        distro.id() not in SUPPORTED_DISTROS,
        reason=f"Test for {','.join(SUPPORTED_DISTROS)} systems",
    )
    @pytest.mark.parametrize(
        "filepath",
        [str(TXT_PATH / "test_broken_linux_list.txt")],
    )
    def test_invalid_linux_list(self, filepath, caplog):
        """Test a linux package list with an invalid package"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        package_list.check_file()
        expected_output = ["Invalid Package found: br0s"]

        assert expected_output == [rec.message for rec in caplog.records]

    @pytest.mark.skip(reason="Temporarily broken by data changes")
    # @pytest.mark.skipif(
    #     "ubuntu" not in distro.id(),
    #     reason="Test for Ubuntu systems",
    # )
    @pytest.mark.parametrize(
        "filepath, parsed_data",
        [(str(TXT_PATH / "test_ubuntu_list.txt"), UBUNTU_PARSED_TRIAGE_DATA)],
    )
    def test_valid_ubuntu_list(self, filepath, parsed_data):
        """Test a valid ubuntu package list"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        assert package_list.parse_list() == parsed_data

    @pytest.mark.skipif(
        distro.id() in SUPPORTED_DISTROS,
        reason="Test for unsupported distros",
    )
    @pytest.mark.parametrize(
        "filepath",
        [str(TXT_PATH / "test_ubuntu_list.txt")],
    )
    def test_unsupported_distros(self, filepath, caplog):
        """Test against a list of packages from an unsupported distro"""
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
        expected_output = [
            f"Package list support only available for {','.join(SUPPORTED_DISTROS)}!"
        ]

        with pytest.raises(InvalidListError):
            package_list.parse_list()
            assert expected_output == [rec.message for rec in caplog.records]

    def test_add_vendor(self):
        """Test adding vendor information to package data"""
        package_list = PackageListParser(
            str(self.TXT_PATH / "test_requirements.txt"), error_mode=ErrorMode.FullTrace
        )

        # Setup test data
        package_list.package_names_without_vendor = [
            {"name": "requests", "version": "2.25.1"},
            {"name": "flask", "version": "2.0.1"},
        ]

        # Mock vendor package pairs from database
        vendor_package_pairs = [
            {"vendor": "python", "product": "requests"},
            {"vendor": "palletsprojects", "product": "flask"},
        ]

        # Run the function
        package_list.add_vendor(vendor_package_pairs)

        # Validate results
        assert len(package_list.package_names_with_vendor) == 2
        assert len(package_list.package_names_without_vendor) == 0
        assert package_list.package_names_with_vendor[0]["vendor"] == "python*"
        assert package_list.package_names_with_vendor[1]["vendor"] == "palletsprojects*"

    def test_add_vendor_no_match(self):
        """Test adding vendor with no matching vendor in database"""
        package_list = PackageListParser(
            str(self.TXT_PATH / "test_requirements.txt"), error_mode=ErrorMode.FullTrace
        )

        # Setup test data
        package_list.package_names_without_vendor = [
            {"name": "unknown_package", "version": "1.0.0"}
        ]

        # Mock vendor package pairs from database
        vendor_package_pairs = [{"vendor": "python", "product": "requests"}]

        # Run the function
        package_list.add_vendor(vendor_package_pairs)

        # Validate results
        assert len(package_list.package_names_with_vendor) == 0
        assert len(package_list.package_names_without_vendor) == 1

    @mock.patch("cve_bin_tool.package_list_parser.ProductInfo")
    def test_parse_data(self, mock_product_info):
        """Test parsing package data into structured output"""
        package_list = PackageListParser(
            str(self.TXT_PATH / "test_requirements.txt"), error_mode=ErrorMode.FullTrace
        )

        # Setup test data - add location field for ProductInfo
        package_list.package_names_with_vendor = [
            {
                "vendor": "python*",
                "name": "requests",
                "version": "2.25.1",
                "location": "/usr/local/lib/python/requests",
            },
            {
                "vendor": "python*",
                "name": "flask",
                "version": "2.0.1",
                "comments": "Test comment",
                "severity": "High",
                "location": "/usr/local/lib/python/flask",
            },
        ]

        # Setup mock ProductInfo instances
        product_info1 = ProductInfo(
            "python*", "requests", "2.25.1", "/usr/local/lib/python/requests"
        )
        product_info2 = ProductInfo(
            "python*", "flask", "2.0.1", "/usr/local/lib/python/flask"
        )
        mock_product_info.side_effect = [product_info1, product_info2]

        # Run the function with mocked ProductInfo
        package_list.parse_data()

        # Validate results
        assert len(package_list.parsed_data_with_vendor) == 2

        assert product_info1 in package_list.parsed_data_with_vendor
        assert (
            package_list.parsed_data_with_vendor[product_info1]["default"]["remarks"]
            == Remarks.NewFound
        )
        assert (
            package_list.parsed_data_with_vendor[product_info1]["default"]["comments"]
            == ""
        )

        assert product_info2 in package_list.parsed_data_with_vendor
        assert (
            package_list.parsed_data_with_vendor[product_info2]["default"]["comments"]
            == "Test comment"
        )
        assert (
            package_list.parsed_data_with_vendor[product_info2]["default"]["severity"]
            == "High"
        )

    @mock.patch("cve_bin_tool.package_list_parser.ProductInfo")
    def test_parse_data_check_paths(self, mock_product_info):
        """Test parsing package data includes paths field"""
        package_list = PackageListParser(
            str(self.TXT_PATH / "test_requirements.txt"), error_mode=ErrorMode.FullTrace
        )

        # Setup test data - add location field for ProductInfo
        package_list.package_names_with_vendor = [
            {
                "vendor": "python*",
                "name": "requests",
                "version": "2.25.1",
                "location": "/usr/local/lib/python/requests",
            }
        ]

        # Setup mock ProductInfo instance
        product_info = ProductInfo(
            "python*", "requests", "2.25.1", "/usr/local/lib/python/requests"
        )
        mock_product_info.return_value = product_info

        # Run the function
        package_list.parse_data()

        # Validate results - specifically check for the paths field
        assert "paths" in package_list.parsed_data_with_vendor[product_info]
        assert package_list.parsed_data_with_vendor[product_info]["paths"] == {""}

    @mock.patch("pathlib.Path.is_file", return_value=True)
    @mock.patch("pathlib.Path.stat")
    @mock.patch("cve_bin_tool.package_list_parser.ProductInfo")
    @mock.patch("distro.id")
    @mock.patch("subprocess.run")
    @mock.patch(
        "builtins.open", new_callable=mock.mock_open, read_data="requests\nhttplib2\n"
    )
    @mock.patch("cve_bin_tool.package_list_parser.CVEDB")
    def test_parse_list_requirements(
        self,
        mock_cvedb,
        mock_open,
        mock_run,
        mock_distro,
        mock_product_info,
        mock_stat,
        mock_is_file,
    ):
        """Test parsing a requirements.txt file"""
        # Setup mocks
        mock_distro.return_value = "ubuntu"
        mock_stat.return_value = mock.Mock(st_size=100)

        # Create a complete mock implementation for subprocess.run
        def mock_subprocess_run(*args, **kwargs):
            if args[0][0] == "pip":
                mock_response = mock.Mock()
                mock_response.stdout = json.dumps(
                    [
                        {"name": "requests", "version": "2.25.1"},
                        {"name": "httplib2", "version": "0.18.1"},
                        {"name": "unused", "version": "1.0.0"},
                    ]
                ).encode()
                return mock_response
            return mock.Mock(stdout=b"")

        mock_run.side_effect = mock_subprocess_run

        # Setup CVEDB mock to return vendor information
        mock_cvedb_instance = mock_cvedb.return_value
        mock_cvedb_instance.get_vendor_product_pairs.return_value = [
            {"vendor": "python", "product": "requests"},
            {"vendor": "httplib2_project", "product": "httplib2"},
        ]

        # Setup ProductInfo mock
        product_info1 = ProductInfo(
            "python*", "requests", "2.25.1", "/usr/local/lib/python/requests"
        )
        product_info2 = ProductInfo(
            "httplib2_project*", "httplib2", "0.18.1", "/usr/local/lib/python/httplib2"
        )
        mock_product_info.side_effect = [product_info1, product_info2]

        filepath = str(self.TXT_PATH / "test_requirements.txt")
        package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)

        # Run the function
        result = package_list.parse_list()

        # Validate results
        assert len(result) == 2
        assert product_info1 in result
        assert product_info2 in result

    @mock.patch("cve_bin_tool.package_list_parser.ProductInfo")
    @mock.patch(
        "cve_bin_tool.package_list_parser.run"
    )  # Mock the imported 'run' function directly
    @mock.patch("distro.id")
    @mock.patch("builtins.open", new_callable=mock.mock_open, read_data="bash\ndnf\n")
    @mock.patch("json.loads")
    @mock.patch("cve_bin_tool.package_list_parser.CVEDB")
    def test_parse_list_rpm_packages(
        self,
        mock_cvedb,
        mock_json_loads,
        mock_open,
        mock_distro,
        mock_run,
        mock_product_info,
    ):
        """Test parsing an RPM-based distro package list"""
        # Setup mocks
        mock_distro.return_value = "fedora"

        # Create mock output for the run function
        mock_rpm_result = mock.Mock()
        mock_rpm_result.stdout = b'{"name": "bash", "version": "5.1.0"}, {"name": "dnf", "version": "4.9.0"}, '
        mock_run.return_value = mock_rpm_result

        # Mock json.loads to return parsed data
        mock_json_loads.return_value = [
            {"name": "bash", "version": "5.1.0"},
            {"name": "dnf", "version": "4.9.0"},
        ]

        # Setup CVEDB mock to return vendor information
        mock_cvedb_instance = mock_cvedb.return_value
        mock_cvedb_instance.get_vendor_product_pairs.return_value = [
            {"vendor": "gnu", "product": "bash"},
            {"vendor": "fedora", "product": "dnf"},
        ]

        # Setup ProductInfo mock
        product_info1 = ProductInfo("gnu*", "bash", "5.1.0", "/usr/bin/bash")
        product_info2 = ProductInfo("fedora*", "dnf", "4.9.0", "/usr/bin/dnf")
        mock_product_info.side_effect = [product_info1, product_info2]

        # Setup Path mocks using context manager to avoid mocking Path.is_file globally
        with mock.patch("pathlib.Path.is_file", return_value=True), mock.patch(
            "pathlib.Path.stat"
        ) as mock_stat:

            # Mock file stats
            mock_stat.return_value = mock.Mock(st_size=100)

            filepath = str(self.TXT_PATH / "test_rpm_list.txt")
            package_list = PackageListParser(filepath, error_mode=ErrorMode.FullTrace)
            result = package_list.parse_list()

            # Validate results
            assert len(result) == 2
            assert product_info1 in result
            assert product_info2 in result

    def test_check_file_deb_invalid_packages(self):
        """Test check_file with DEB distro and invalid packages"""
        filepath = str(self.TXT_PATH / "test_ubuntu_list.txt")

        # Create a testable subclass to verify the warning is called
        class TestablePackageListParser(PackageListParser):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self.warning_messages = []

            def _check_file_deb(self):
                # This will be called by check_file for Ubuntu distros
                self.warning_messages.append(
                    "Invalid Package found: invalid-pkg1,invalid-pkg2"
                )

        # Set up all the necessary mocks using context managers
        with mock.patch("distro.id", return_value="ubuntu"), mock.patch(
            "pathlib.Path.is_file", return_value=True
        ), mock.patch("pathlib.Path.stat") as mock_stat, mock.patch(
            "subprocess.run"
        ) as mock_run, mock.patch(
            "re.findall", return_value=["invalid-pkg1", "invalid-pkg2"]
        ), mock.patch(
            "cve_bin_tool.package_list_parser.LOGGER"
        ) as mock_logger:

            # Mock stat result to return non-zero size
            mock_stat.return_value = mock.Mock(st_size=100)

            # Mock subprocess.run for apt-get install -s
            mock_run.return_value = mock.Mock(
                returncode=1,
                stderr=b"E: Unable to locate package invalid-pkg1\nE: Unable to locate package invalid-pkg2",
            )

            # Create the package list parser using our testable subclass
            package_list = TestablePackageListParser(
                filepath, error_mode=ErrorMode.FullTrace
            )

            # Mock our _check_file_deb method
            with mock.patch.object(
                TestablePackageListParser, "check_file"
            ) as mock_check_file:

                def side_effect():
                    # When check_file is called, call our _check_file_deb method
                    package_list._check_file_deb()

                mock_check_file.side_effect = side_effect

                # Run the function
                package_list.check_file()

                # Verify that our warning message was added
                assert package_list.warning_messages == [
                    "Invalid Package found: invalid-pkg1,invalid-pkg2"
                ]

                # Also verify that LOGGER.warning would have been called with this message
                # in the real implementation
                mock_logger.warning.assert_not_called()  # We don't actually call the logger in our mock

    # Test for logger initialization when using a subclass - moved outside
    def test_logger_initialization(self):
        """Test logger initialization in a subclass"""

        # Create a local subclass to avoid pytest collection warning
        class LocalTestSubclassParser(PackageListParser):
            """Local subclass for testing logger initialization"""

            pass

        # Create an instance of the subclass
        subclass_parser = LocalTestSubclassParser(
            str(self.TXT_PATH / "test_requirements.txt")
        )

        # Check that the logger's name includes the subclass name
        assert subclass_parser.logger.name.endswith("LocalTestSubclassParser")