WIP: Typing changes left

Author: DanielNoord

pylint/checkers/similar.py

@@ -49,10 +49,11 @@
 import sys
 from collections import defaultdict
 from getopt import getopt
-from io import BufferedIOBase, BufferedReader, BytesIO
+from io import BufferedReader, BytesIO
 from itertools import chain, groupby
 from typing import (
     Any,
+    Callable,
     Dict,
     FrozenSet,
     Generator,
@@ -379,12 +380,10 @@ def append_stream(
         self, streamid: str, stream: STREAM_TYPES, encoding: Optional[str] = None
     ) -> None:
         """append a file to search for similarities"""
-        if isinstance(stream, BufferedIOBase):
-            if encoding is None:
-                raise ValueError
-            readlines = decoding_stream(stream, encoding).readlines
+        if encoding is None:
+            readlines: Union[Callable[..., Union[str, bytes]], Any] = stream.readlines
         else:
-            readlines = stream.readlines  # type: ignore # hint parameter is incorrectly typed as non-optional
+            readlines = decoding_stream(stream, encoding).readlines
         try:
             self.linesets.append(
                 LineSet(

pylint/reporters/base_reporter.py

@@ -21,7 +21,7 @@ class BaseReporter:
 
     extension = ""
 
-    def __init__(self, output=None):
+    def __init__(self, output=None) -> None:
         self.linter = None
         self.section = 0
         self.out = None
@@ -39,7 +39,7 @@ def set_output(self, output=None):
         """set output stream"""
         self.out = output or sys.stdout
 
-    def writeln(self, string=""):
+    def writeln(self, string: str = "") -> None:
         """write a line in the output buffer"""
         print(string, file=self.out)
 

pylint/reporters/text.py

@@ -26,12 +26,14 @@
 import os
 import sys
 import warnings
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Set
 
 from pylint import utils
 from pylint.interfaces import IReporter
+from pylint.lint.pylinter import PyLinter
 from pylint.message import Message
 from pylint.reporters import BaseReporter
+from pylint.reporters.ureports.nodes import Section
 from pylint.reporters.ureports.text_writer import TextWriter
 
 if TYPE_CHECKING:
@@ -64,7 +66,7 @@
 }
 
 
-def _get_ansi_code(color=None, style=None):
+def _get_ansi_code(color: Optional[str] = None, style: Optional[str] = None) -> str:
     """return ansi escape code corresponding to color and style
 
     :type color: str or None
@@ -98,7 +100,7 @@ def _get_ansi_code(color=None, style=None):
     return ""
 
 
-def colorize_ansi(msg, color=None, style=None):
+def colorize_ansi(msg: str, color: str = None, style: Optional[str] = None) -> str:
     """colorize message by wrapping it with ansi escape codes
 
     :type msg: str or unicode
@@ -136,15 +138,15 @@ class TextReporter(BaseReporter):
     extension = "txt"
     line_format = "{path}:{line}:{column}: {msg_id}: {msg} ({symbol})"
 
-    def __init__(self, output=None):
+    def __init__(self, output: Optional[StringIO] = None) -> None:
         BaseReporter.__init__(self, output)
-        self._modules = set()
+        self._modules: Set[str] = set()
         self._template = self.line_format
 
     def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
         self._template = str(self.linter.config.msg_template or self._template)
 
-    def write_message(self, msg):
+    def write_message(self, msg: Message) -> None:
         """Convenience method to write a formatted message with class default template"""
         self.writeln(msg.format(self._template))
 
@@ -174,7 +176,7 @@ class ParseableTextReporter(TextReporter):
     name = "parseable"
     line_format = "{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}"
 
-    def __init__(self, output=None):
+    def __init__(self, output: Optional[Any] = None) -> None:
         warnings.warn(
             f"{self.name} output format is deprecated. This is equivalent to --msg-template={self.line_format}",
             DeprecationWarning,
@@ -203,7 +205,9 @@ class ColorizedTextReporter(TextReporter):
         "S": ("yellow", "inverse"),  # S stands for module Separator
     }
 
-    def __init__(self, output=None, color_mapping=None):
+    def __init__(
+        self, output: Optional[StringIO] = None, color_mapping: Optional[Any] = None
+    ) -> None:
         TextReporter.__init__(self, output)
         self.color_mapping = color_mapping or dict(ColorizedTextReporter.COLOR_MAPPING)
         ansi_terms = ["xterm-16color", "xterm-256color"]
@@ -214,7 +218,9 @@ def __init__(self, output=None, color_mapping=None):
 
                 self.out = colorama.AnsiToWin32(self.out)
 
-    def _get_decoration(self, msg_id):
+    def _get_decoration(
+        self, msg_id: str
+    ) -> Union[Tuple[None, None], Tuple[str, Optional[str]]]:
         """Returns the tuple color, style associated with msg_id as defined
         in self.color_mapping
         """
@@ -248,7 +254,7 @@ def handle_message(self, msg: Message) -> None:
         self.write_message(msg)
 
 
-def register(linter):
+def register(linter: PyLinter) -> None:
     """Register the reporter classes with the linter."""
     linter.register_reporter(TextReporter)
     linter.register_reporter(ParseableTextReporter)

pylint/testutils/checker_test_case.py

@@ -2,34 +2,37 @@
 # For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
 
 import contextlib
-from typing import Dict, Optional, Type
+from typing import Dict, Iterator, Type
+
+from astroid.nodes import Module
 
 from pylint.testutils.global_test_linter import linter
+from pylint.testutils.output_line import Message
 from pylint.testutils.unittest_linter import UnittestLinter
 from pylint.utils import ASTWalker
 
 
 class CheckerTestCase:
     """A base testcase class for unit testing individual checker classes."""
 
-    CHECKER_CLASS: Optional[Type] = None
+    CHECKER_CLASS: Type
     CONFIG: Dict = {}
 
-    def setup_method(self):
+    def setup_method(self) -> None:
         self.linter = UnittestLinter()
         self.checker = self.CHECKER_CLASS(self.linter)  # pylint: disable=not-callable
         for key, value in self.CONFIG.items():
             setattr(self.checker.config, key, value)
         self.checker.open()
 
     @contextlib.contextmanager
-    def assertNoMessages(self):
+    def assertNoMessages(self) -> Iterator[None]:
         """Assert that no messages are added by the given method."""
         with self.assertAddsMessages():
             yield
 
     @contextlib.contextmanager
-    def assertAddsMessages(self, *messages):
+    def assertAddsMessages(self, *messages: Message) -> Iterator[None]:
         """Assert that exactly the given method adds the given messages.
 
         The list of messages must exactly match *all* the messages added by the
@@ -47,7 +50,7 @@ def assertAddsMessages(self, *messages):
         )
         assert got == list(messages), msg
 
-    def walk(self, node):
+    def walk(self, node: Module) -> None:
         """recursive walk on the given node"""
         walker = ASTWalker(linter)
         walker.add_checker(self.checker)

pylint/testutils/lint_module_test.py

@@ -7,7 +7,7 @@
 import sys
 from collections import Counter
 from io import StringIO, TextIOWrapper
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Optional, TextIO, Tuple, Union
 
 import pytest
 from _pytest.config import Config
@@ -53,7 +53,7 @@ def __init__(self, test_file: FunctionalTestFile, config: Optional[Config] = Non
         self._test_file = test_file
         self._config = config
 
-    def setUp(self):
+    def setUp(self) -> None:
         if self._should_be_skipped_due_to_version():
             pytest.skip(
                 f"Test cannot run with Python {sys.version.split(' ', maxsplit=1)[0]}."
@@ -78,10 +78,10 @@ def setUp(self):
             if sys.platform.lower() in platforms:
                 pytest.skip(f"Test cannot run on platform {sys.platform!r}")
 
-    def runTest(self):
+    def runTest(self) -> None:
         self._runTest()
 
-    def _should_be_skipped_due_to_version(self):
+    def _should_be_skipped_due_to_version(self) -> bool:
         return (
             sys.version_info < self._test_file.options["min_pyver"]
             or sys.version_info > self._test_file.options["max_pyver"]
@@ -141,21 +141,21 @@ def multiset_difference(
         return missing, unexpected
 
     # pylint: disable=consider-using-with
-    def _open_expected_file(self):
+    def _open_expected_file(self) -> Union[StringIO, TextIOWrapper, TextIO]:
         try:
             return open(self._test_file.expected_output, encoding="utf-8")
         except FileNotFoundError:
             return StringIO("")
 
     # pylint: disable=consider-using-with
-    def _open_source_file(self):
+    def _open_source_file(self) -> TextIO:
         if self._test_file.base == "invalid_encoded_data":
             return open(self._test_file.source, encoding="utf-8")
         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):
+    def _get_expected(self) -> Tuple[Counter, List[OutputLine]]:
         with self._open_source_file() as f:
             expected_msgs = self.get_expected_messages(f)
         if not expected_msgs:
@@ -166,10 +166,10 @@ def _get_expected(self):
             ]
         return expected_msgs, expected_output_lines
 
-    def _get_actual(self):
+    def _get_actual(self) -> Tuple[Counter, List[OutputLine]]:
         messages = self._linter.reporter.messages
         messages.sort(key=lambda m: (m.line, m.symbol, m.msg))
-        received_msgs = Counter()
+        received_msgs: Counter = Counter()
         received_output_lines = []
         for msg in messages:
             assert (
@@ -179,7 +179,7 @@ def _get_actual(self):
             received_output_lines.append(OutputLine.from_msg(msg))
         return received_msgs, received_output_lines
 
-    def _runTest(self):
+    def _runTest(self) -> None:
         __tracebackhide__ = True  # pylint: disable=unused-variable
         modules_to_check = [self._test_file.source]
         self._linter.check(modules_to_check)
@@ -234,7 +234,12 @@ def error_msg_for_unequal_output(self, expected_lines, received_lines) -> str:
                 error_msg += f"{line}\n"
         return error_msg
 
-    def _check_output_text(self, _, expected_output, actual_output):
+    def _check_output_text(
+        self,
+        _: Counter,
+        expected_output: List[OutputLine],
+        actual_output: List[OutputLine],
+    ) -> None:
         """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

pylint/testutils/output_line.py

@@ -2,10 +2,11 @@
 # For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
 
 import collections
-from typing import Any, NamedTuple
+from typing import List, NamedTuple, Tuple, Union
 
 from pylint import interfaces
 from pylint.constants import PY38_PLUS
+from pylint.message.message import Message as pylint_Message
 from pylint.testutils.constants import UPDATE_OPTION
 
 
@@ -15,16 +16,20 @@ class Message(
     def __new__(cls, msg_id, line=None, node=None, args=None, confidence=None):
         return tuple.__new__(cls, (msg_id, line, node, args, confidence))
 
-    def __eq__(self, other):
+    def __eq__(self, other: Union[object, "Message"]) -> bool:
         if isinstance(other, Message):
             if self.confidence and other.confidence:
                 return super().__eq__(other)
-            return self[:-1] == other[:-1]
+            return self[:-1] == other[:-1]  # type: ignore
         return NotImplemented  # pragma: no cover
 
 
 class MalformedOutputLineException(Exception):
-    def __init__(self, row, exception):
+    def __init__(
+        self,
+        row: Union[List[str], Tuple[str, str, str, str, str], str],
+        exception: ValueError,
+    ) -> None:
         example = "msg-symbolic-name:42:27:MyClass.my_function:The message"
         other_example = "msg-symbolic-name:7:42::The message"
         expected = [
@@ -59,12 +64,12 @@ class OutputLine(NamedTuple):
     symbol: str
     lineno: int
     column: str
-    object: Any
+    object: str
     msg: str
-    confidence: interfaces.Confidence
+    confidence: Union[interfaces.Confidence, str]
 
     @classmethod
-    def from_msg(cls, msg):
+    def from_msg(cls, msg: pylint_Message) -> "OutputLine":
         column = cls.get_column(msg.column)
         return cls(
             msg.symbol,
@@ -78,19 +83,26 @@ def from_msg(cls, msg):
         )
 
     @classmethod
-    def get_column(cls, column):
+    def get_column(cls, column: Union[int, str]) -> str:
         if not PY38_PLUS:
             return ""  # pragma: no cover
         return str(column)
 
     @classmethod
-    def from_csv(cls, row):
+    def from_csv(
+        cls, row: Union[List[str], Tuple[str, str, str, str, str], str]
+    ) -> "OutputLine":
         try:
-            confidence = row[5] if len(row) == 6 else interfaces.HIGH.name
+            if len(row) == 6:
+                confidence = row[5]  # type: ignore # mypy does not recognize that this cannot fail
+            else:
+                confidence = interfaces.HIGH.name
             column = cls.get_column(row[2])
             return cls(row[0], int(row[1]), column, row[3], row[4], confidence)
         except Exception as e:
-            raise MalformedOutputLineException(row, e) from e
+            if isinstance(e, ValueError):
+                raise MalformedOutputLineException(row, e) from e
+            raise
 
-    def to_csv(self):
-        return tuple(self)
+    def to_csv(self) -> "OutputLine":
+        return self

pylint/testutils/reporter_for_tests.py

@@ -23,7 +23,7 @@ def __init__(
     ):  # pylint: disable=super-init-not-called # See https://github.com/PyCQA/pylint/issues/4941
         self.reset()
 
-    def reset(self):
+    def reset(self) -> None:
         self.message_ids: Dict = {}
         self.out = StringIO()
         self.path_strip_prefix: str = getcwd() + sep
@@ -45,11 +45,12 @@ def handle_message(self, msg: Message) -> None:
             str_message = str_message.replace("\r\n", "\n")
         self.messages.append(f"{sigle}:{line:>3}{obj}: {str_message}")
 
-    def finalize(self):
+    def finalize(self) -> str:
         self.messages.sort()
         for msg in self.messages:
             print(msg, file=self.out)
-        result = self.out.getvalue()
+        if isinstance(self.out, StringIO):
+            result = self.out.getvalue()
         self.reset()
         return result