Avoid false unreachable, redundant-expr, and redundant-casts warnings in loops more robustly and efficiently, and avoid multiple revealed type notes for the same line. (#19118)

Fixes #18606 
Closes #18511
Improves #18991
Fixes #19170

This change is an improvement over 9685171. Besides fixing the
regressions reported in #18606 and #19170 and removing the duplicates
reported in #18511, it should significantly reduce the performance
regression reported in #18991. At least running `Measure-command {python
runtests.py self}` on my computer (with removed cache) is 10 % faster.

Author: tyralla

mypy/checker.py

@@ -25,7 +25,7 @@
 from mypy.constraints import SUPERTYPE_OF
 from mypy.erasetype import erase_type, erase_typevars, remove_instance_last_known_values
 from mypy.errorcodes import TYPE_VAR, UNUSED_AWAITABLE, UNUSED_COROUTINE, ErrorCode
-from mypy.errors import Errors, ErrorWatcher, report_internal_error
+from mypy.errors import Errors, ErrorWatcher, LoopErrorWatcher, report_internal_error
 from mypy.expandtype import expand_type
 from mypy.literals import Key, extract_var_from_literal_hash, literal, literal_hash
 from mypy.maptype import map_instance_to_supertype
@@ -599,19 +599,27 @@ def accept_loop(
             # on without bound otherwise)
             widened_old = len(self.widened_vars)
 
-            # Disable error types that we cannot safely identify in intermediate iteration steps:
-            warn_unreachable = self.options.warn_unreachable
-            warn_redundant = codes.REDUNDANT_EXPR in self.options.enabled_error_codes
-            self.options.warn_unreachable = False
-            self.options.enabled_error_codes.discard(codes.REDUNDANT_EXPR)
-
+            # one set of `unreachable`, `redundant-expr`, and `redundant-casts` errors
+            # per iteration step:
+            uselessness_errors = []
+            # one set of unreachable line numbers per iteration step:
+            unreachable_lines = []
+            # one set of revealed types per line where `reveal_type` is used (each
+            # created set can grow during the iteration):
+            revealed_types = defaultdict(set)
             iter = 1
             while True:
                 with self.binder.frame_context(can_skip=True, break_frame=2, continue_frame=1):
                     if on_enter_body is not None:
                         on_enter_body()
 
-                    self.accept(body)
+                    with LoopErrorWatcher(self.msg.errors) as watcher:
+                        self.accept(body)
+                    uselessness_errors.append(watcher.uselessness_errors)
+                    unreachable_lines.append(watcher.unreachable_lines)
+                    for key, values in watcher.revealed_types.items():
+                        revealed_types[key].update(values)
+
                 partials_new = sum(len(pts.map) for pts in self.partial_types)
                 widened_new = len(self.widened_vars)
                 # Perform multiple iterations if something changed that might affect
@@ -632,16 +640,29 @@ def accept_loop(
                 if iter == 20:
                     raise RuntimeError("Too many iterations when checking a loop")
 
-            # If necessary, reset the modified options and make up for the postponed error checks:
-            self.options.warn_unreachable = warn_unreachable
-            if warn_redundant:
-                self.options.enabled_error_codes.add(codes.REDUNDANT_EXPR)
-            if warn_unreachable or warn_redundant:
-                with self.binder.frame_context(can_skip=True, break_frame=2, continue_frame=1):
-                    if on_enter_body is not None:
-                        on_enter_body()
-
-                    self.accept(body)
+            # Report only those `unreachable`, `redundant-expr`, and `redundant-casts`
+            # errors that could not be ruled out in any iteration step:
+            persistent_uselessness_errors = set()
+            for candidate in set(itertools.chain(*uselessness_errors)):
+                if all(
+                    (candidate in errors) or (candidate[2] in lines)
+                    for errors, lines in zip(uselessness_errors, unreachable_lines)
+                ):
+                    persistent_uselessness_errors.add(candidate)
+            for error_info in persistent_uselessness_errors:
+                context = Context(line=error_info[2], column=error_info[3])
+                context.end_line = error_info[4]
+                context.end_column = error_info[5]
+                self.msg.fail(error_info[1], context, code=error_info[0])
+
+            #  Report all types revealed in at least one iteration step:
+            for note_info, types in revealed_types.items():
+                sorted_ = sorted(types, key=lambda typ: typ.lower())
+                revealed = sorted_[0] if len(types) == 1 else f"Union[{', '.join(sorted_)}]"
+                context = Context(line=note_info[1], column=note_info[2])
+                context.end_line = note_info[3]
+                context.end_column = note_info[4]
+                self.note(f'Revealed type is "{revealed}"', context)
 
             # If exit_condition is set, assume it must be False on exit from the loop:
             if exit_condition:

mypy/errors.py

@@ -6,7 +6,7 @@
 from collections import defaultdict
 from collections.abc import Iterable
 from typing import Callable, Final, NoReturn, Optional, TextIO, TypeVar
-from typing_extensions import Literal, TypeAlias as _TypeAlias
+from typing_extensions import Literal, Self, TypeAlias as _TypeAlias
 
 from mypy import errorcodes as codes
 from mypy.error_formatter import ErrorFormatter
@@ -179,7 +179,7 @@ def __init__(
         self._filter_deprecated = filter_deprecated
         self._filtered: list[ErrorInfo] | None = [] if save_filtered_errors else None
 
-    def __enter__(self) -> ErrorWatcher:
+    def __enter__(self) -> Self:
         self.errors._watchers.append(self)
         return self
 
@@ -220,6 +220,60 @@ def filtered_errors(self) -> list[ErrorInfo]:
         return self._filtered
 
 
+class LoopErrorWatcher(ErrorWatcher):
+    """Error watcher that filters and separately collects `unreachable` errors,
+    `redundant-expr` and `redundant-casts` errors, and revealed types when analysing
+    loops iteratively to help avoid making too-hasty reports."""
+
+    # Meaning of the tuple items: ErrorCode, message, line, column, end_line, end_column:
+    uselessness_errors: set[tuple[ErrorCode, str, int, int, int, int]]
+
+    # Meaning of the tuple items: function_or_member, line, column, end_line, end_column:
+    revealed_types: dict[tuple[str | None, int, int, int, int], set[str]]
+
+    # Not only the lines where the error report occurs but really all unreachable lines:
+    unreachable_lines: set[int]
+
+    def __init__(
+        self,
+        errors: Errors,
+        *,
+        filter_errors: bool | Callable[[str, ErrorInfo], bool] = False,
+        save_filtered_errors: bool = False,
+        filter_deprecated: bool = False,
+    ) -> None:
+        super().__init__(
+            errors,
+            filter_errors=filter_errors,
+            save_filtered_errors=save_filtered_errors,
+            filter_deprecated=filter_deprecated,
+        )
+        self.uselessness_errors = set()
+        self.unreachable_lines = set()
+        self.revealed_types = defaultdict(set)
+
+    def on_error(self, file: str, info: ErrorInfo) -> bool:
+
+        if info.code in (codes.UNREACHABLE, codes.REDUNDANT_EXPR, codes.REDUNDANT_CAST):
+            self.uselessness_errors.add(
+                (info.code, info.message, info.line, info.column, info.end_line, info.end_column)
+            )
+            if info.code == codes.UNREACHABLE:
+                self.unreachable_lines.update(range(info.line, info.end_line + 1))
+            return True
+
+        if info.code == codes.MISC and info.message.startswith("Revealed type is "):
+            key = info.function_or_member, info.line, info.column, info.end_line, info.end_column
+            types = info.message.split('"')[1]
+            if types.startswith("Union["):
+                self.revealed_types[key].update(types[6:-1].split(", "))
+            else:
+                self.revealed_types[key].add(types)
+            return True
+
+        return super().on_error(file, info)
+
+
 class Errors:
     """Container for compile errors.
 

test-data/unit/check-inference.test

@@ -343,7 +343,7 @@ for var2 in [g, h, i, j, k, l]:
     reveal_type(var2)  # N: Revealed type is "Union[builtins.int, builtins.str]"
 
 for var3 in [m, n, o, p, q, r]:
-    reveal_type(var3)  # N: Revealed type is "Union[builtins.int, Any]"
+    reveal_type(var3)  # N: Revealed type is "Union[Any, builtins.int]"
 
 T = TypeVar("T", bound=Type[Foo])
 
@@ -1247,7 +1247,7 @@ class X(TypedDict):
 
 x: X
 for a in ("hourly", "daily"):
-    reveal_type(a)  # N: Revealed type is "Union[Literal['hourly']?, Literal['daily']?]"
+    reveal_type(a)  # N: Revealed type is "Union[Literal['daily']?, Literal['hourly']?]"
     reveal_type(x[a])  # N: Revealed type is "builtins.int"
     reveal_type(a.upper())  # N: Revealed type is "builtins.str"
     c = a

test-data/unit/check-narrowing.test

@@ -2346,8 +2346,7 @@ def f() -> bool: ...
 
 y = None
 while f():
-    reveal_type(y)  # N: Revealed type is "None" \
-                    # N: Revealed type is "Union[builtins.int, None]"
+    reveal_type(y)  # N: Revealed type is "Union[builtins.int, None]"
     y = 1
 reveal_type(y)  # N: Revealed type is "Union[builtins.int, None]"
 
@@ -2370,7 +2369,42 @@ class A:
 
 [builtins fixtures/primitives.pyi]
 
-[case testAvoidFalseUnreachableInLoop]
+[case testPersistentUnreachableLinesNestedInInpersistentUnreachableLines]
+# flags: --warn-unreachable --python-version 3.11
+
+x = None
+y = None
+while True:
+    if x is not None:
+        if y is not None:
+            reveal_type(y)  # E: Statement is unreachable
+    x = 1
+
+[builtins fixtures/bool.pyi]
+
+[case testAvoidFalseRedundantCastInLoops]
+# flags: --warn-redundant-casts
+
+from typing import Callable, cast, Union
+
+ProcessorReturnValue = Union[str, int]
+Processor = Callable[[str], ProcessorReturnValue]
+
+def main_cast(p: Processor) -> None:
+    ed: ProcessorReturnValue
+    ed = cast(str, ...)
+    while True:
+        ed = p(cast(str, ed))
+
+def main_no_cast(p: Processor) -> None:
+    ed: ProcessorReturnValue
+    ed = cast(str, ...)
+    while True:
+        ed = p(ed)  # E: Argument 1 has incompatible type "Union[str, int]"; expected "str"
+
+[builtins fixtures/bool.pyi]
+
+[case testAvoidFalseUnreachableInLoop1]
 # flags: --warn-unreachable --python-version 3.11
 
 def f() -> int | None: ...
@@ -2383,6 +2417,29 @@ while x is not None or b():
 
 [builtins fixtures/bool.pyi]
 
+[case testAvoidFalseUnreachableInLoop2]
+# flags: --warn-unreachable --python-version 3.11
+
+y = None
+while y is None:
+    if y is None:
+        y = []
+    y.append(1)
+
+[builtins fixtures/list.pyi]
+
+[case testAvoidFalseUnreachableInLoop3]
+# flags: --warn-unreachable --python-version 3.11
+
+xs: list[int | None]
+y = None
+for x in xs:
+    if x is not None:
+        if y is None:
+            y = {}  # E: Need type annotation for "y" (hint: "y: Dict[<type>, <type>] = ...")
+
+[builtins fixtures/list.pyi]
+
 [case testAvoidFalseRedundantExprInLoop]
 # flags: --enable-error-code redundant-expr --python-version 3.11
 

test-data/unit/check-redefine2.test

@@ -628,8 +628,7 @@ def f1() -> None:
 def f2() -> None:
     x = None
     while int():
-        reveal_type(x) # N: Revealed type is "None" \
-                       # N: Revealed type is "Union[None, builtins.str]"
+        reveal_type(x) # N: Revealed type is "Union[builtins.str, None]"
         if int():
             x = ""
     reveal_type(x) # N: Revealed type is "Union[None, builtins.str]"
@@ -709,8 +708,7 @@ def b() -> None:
 def c() -> None:
     x = 0
     while int():
-        reveal_type(x) # N: Revealed type is "builtins.int" \
-                       # N: Revealed type is "Union[builtins.int, builtins.str, None]"
+        reveal_type(x)  # N: Revealed type is "Union[builtins.int, builtins.str, None]"
         if int():
             x = ""
             continue
@@ -810,8 +808,7 @@ def f4() -> None:
                         x = None
                         break
         finally:
-            reveal_type(x) # N: Revealed type is "Union[builtins.int, builtins.str, None]" \
-                # N: Revealed type is "Union[builtins.int, None]"
+            reveal_type(x) # N: Revealed type is "Union[builtins.int, builtins.str, None]"
         reveal_type(x) # N: Revealed type is "Union[builtins.int, None]"
 [builtins fixtures/exception.pyi]
 
@@ -927,7 +924,7 @@ class X(TypedDict):
 
 x: X
 for a in ("hourly", "daily"):
-    reveal_type(a)  # N: Revealed type is "Union[Literal['hourly']?, Literal['daily']?]"
+    reveal_type(a)  # N: Revealed type is "Union[Literal['daily']?, Literal['hourly']?]"
     reveal_type(x[a])  # N: Revealed type is "builtins.int"
     reveal_type(a.upper())  # N: Revealed type is "builtins.str"
     c = a

test-data/unit/check-typevar-tuple.test

@@ -989,7 +989,7 @@ from typing_extensions import Unpack
 
 def pipeline(*xs: Unpack[Tuple[int, Unpack[Tuple[float, ...]], bool]]) -> None:
     for x in xs:
-        reveal_type(x)  # N: Revealed type is "Union[builtins.int, builtins.float]"
+        reveal_type(x)  # N: Revealed type is "Union[builtins.float, builtins.int]"
 [builtins fixtures/tuple.pyi]
 
 [case testFixedUnpackItemInInstanceArguments]