lint_module_test.py

# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/master/LICENSE

import csv
import operator
import platform
import sys
from collections import Counter
from io import StringIO
from typing import Dict, List, Optional, Tuple

import pytest
from _pytest.config import Config

from pylint import checkers
from pylint.lint import PyLinter
from pylint.testutils.constants import _EXPECTED_RE, _OPERATORS, UPDATE_OPTION
from pylint.testutils.functional_test_file import (
    FunctionalTestFile,
    NoFileError,
    parse_python_version,
)
from pylint.testutils.output_line import OutputLine
from pylint.testutils.reporter_for_tests import FunctionalTestReporter
from pylint.utils import utils


class LintModuleTest:
    maxDiff = None

    def __init__(self, test_file: FunctionalTestFile, config: Optional[Config] = None):
        _test_reporter = FunctionalTestReporter()
        self._linter = PyLinter()
        self._linter.set_reporter(_test_reporter)
        self._linter.config.persistent = 0
        checkers.initialize(self._linter)
        self._linter.disable("suppressed-message")
        self._linter.disable("locally-disabled")
        self._linter.disable("useless-suppression")
        try:
            self._linter.read_config_file(test_file.option_file)
            if self._linter.cfgfile_parser.has_option("MASTER", "load-plugins"):
                plugins = utils._splitstrip(
                    self._linter.cfgfile_parser.get("MASTER", "load-plugins")
                )
                self._linter.load_plugin_modules(plugins)
            self._linter.load_config_file()
        except NoFileError:
            pass
        self._test_file = test_file
        self._config = config

    def setUp(self):
        if self._should_be_skipped_due_to_version():
            pytest.skip(
                "Test cannot run with Python %s." % sys.version.partition(" ")[0]
            )
        missing = []
        for requirement in self._test_file.options["requires"]:
            try:
                __import__(requirement)
            except ImportError:
                missing.append(requirement)
        if missing:
            pytest.skip("Requires %s to be present." % ",".join(missing))
        except_implementations = self._test_file.options["except_implementations"]
        if except_implementations:
            implementations = [i.strip() for i in except_implementations.split(",")]
            if platform.python_implementation() in implementations:
                msg = "Test cannot run with Python implementation %r"
                pytest.skip(msg % platform.python_implementation())
        excluded_platforms = self._test_file.options["exclude_platforms"]
        if excluded_platforms:
            platforms = [p.strip() for p in excluded_platforms.split(",")]
            if sys.platform.lower() in platforms:
                pytest.skip("Test cannot run on platform %r" % sys.platform)

    def _should_be_skipped_due_to_version(self):
        return (
            sys.version_info < self._test_file.options["min_pyver"]
            or sys.version_info > self._test_file.options["max_pyver"]
        )

    def __str__(self):
        return f"{self._test_file.base} ({self.__class__.__module__}.{self.__class__.__name__})"

    @staticmethod
    def get_expected_messages(stream):
        """Parses a file and get expected messages.

        :param stream: File-like input stream.
        :type stream: enumerable
        :returns: A dict mapping line,msg-symbol tuples to the count on this line.
        :rtype: dict
        """
        messages = Counter()
        for i, line in enumerate(stream):
            match = _EXPECTED_RE.search(line)
            if match is None:
                continue
            line = match.group("line")
            if line is None:
                line = i + 1
            elif line.startswith("+") or line.startswith("-"):
                line = i + 1 + int(line)
            else:
                line = int(line)

            version = match.group("version")
            op = match.group("op")
            if version:
                required = parse_python_version(version)
                if not _OPERATORS[op](sys.version_info, required):
                    continue

            for msg_id in match.group("msgs").split(","):
                messages[line, msg_id.strip()] += 1
        return messages

    @staticmethod
    def multiset_difference(
        expected_entries: Counter, actual_entries: Counter
    ) -> Tuple[Counter, Dict[str, int]]:
        """Takes two multisets and compares them.

        A multiset is a dict with the cardinality of the key as the value."""
        missing = expected_entries.copy()
        missing.subtract(actual_entries)
        unexpected = {}
        for key, value in list(missing.items()):
            if value <= 0:
                missing.pop(key)
                if value < 0:
                    unexpected[key] = -value
        return missing, unexpected

    # pylint: disable=consider-using-with
    def _open_expected_file(self):
        try:
            return open(self._test_file.expected_output)
        except FileNotFoundError:
            return StringIO("")

    # pylint: disable=consider-using-with
    def _open_source_file(self):
        if self._test_file.base == "invalid_encoded_data":
            return open(self._test_file.source)
        if "latin1" in self._test_file.base:
            return open(self._test_file.source, encoding="latin1")
        return open(self._test_file.source, encoding="utf8")

    def _get_expected(self):
        with self._open_source_file() as f:
            expected_msgs = self.get_expected_messages(f)
        if not expected_msgs:
            return Counter(), []
        with self._open_expected_file() as f:
            expected_output_lines = [
                OutputLine.from_csv(row) for row in csv.reader(f, "test")
            ]
        return expected_msgs, expected_output_lines

    def _get_actual(self):
        messages = self._linter.reporter.messages
        messages.sort(key=lambda m: (m.line, m.symbol, m.msg))
        received_msgs = Counter()
        received_output_lines = []
        for msg in messages:
            assert (
                msg.symbol != "fatal"
            ), f"Pylint analysis failed because of '{msg.msg}'"
            received_msgs[msg.line, msg.symbol] += 1
            received_output_lines.append(OutputLine.from_msg(msg))
        return received_msgs, received_output_lines

    def _runTest(self):
        modules_to_check = [self._test_file.source]
        self._linter.check(modules_to_check)
        expected_messages, expected_output = self._get_expected()
        actual_messages, actual_output = self._get_actual()
        assert (
            expected_messages == actual_messages
        ), self.error_msg_for_unequal_messages(
            actual_messages, expected_messages, actual_output
        )
        self._check_output_text(expected_messages, expected_output, actual_output)

    def error_msg_for_unequal_messages(
        self, actual_messages, expected_messages, actual_output: List[OutputLine]
    ):
        msg = ['Wrong results for file "%s":' % (self._test_file.base)]
        missing, unexpected = self.multiset_difference(
            expected_messages, actual_messages
        )
        if missing:
            msg.append("\nExpected in testdata:")
            msg.extend(" %3d: %s" % msg for msg in sorted(missing))
        if unexpected:
            msg.append("\nUnexpected in testdata:")
            msg.extend(" %3d: %s" % msg for msg in sorted(unexpected))  # type: ignore
        error_msg = "\n".join(msg)
        if self._config and self._config.getoption("verbose") > 0:
            error_msg += "\n\nActual pylint output for this file:\n"
            error_msg += "\n".join(str(o) for o in actual_output)
        return error_msg

    def error_msg_for_unequal_output(self, expected_lines, received_lines) -> str:
        missing = set(expected_lines) - set(received_lines)
        unexpected = set(received_lines) - set(expected_lines)
        error_msg = (
            f"Wrong output for '{self._test_file.base}.txt':\n"
            "You can update the expected output automatically with: '"
            f"python tests/test_functional.py {UPDATE_OPTION} -k "
            f'"test_functional[{self._test_file.base}]"\'\n\n'
        )
        sort_by_line_number = operator.attrgetter("lineno")
        if missing:
            error_msg += "\n- Missing lines:\n"
            for line in sorted(missing, key=sort_by_line_number):
                error_msg += f"{line}\n"
        if unexpected:
            error_msg += "\n- Unexpected lines:\n"
            for line in sorted(unexpected, key=sort_by_line_number):
                error_msg += f"{line}\n"
        return error_msg

    def _check_output_text(self, _, expected_output, actual_output):
        """This is a function because we want to be able to update the text in LintModuleOutputUpdate"""
        assert expected_output == actual_output, self.error_msg_for_unequal_output(
            expected_output, actual_output
        )