Improve the handling of "iteration dependent" errors and notes in finally clauses.

mypy/checker.py

@@ -25,7 +25,14 @@
 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 ErrorInfo, Errors, ErrorWatcher, LoopErrorWatcher, report_internal_error
+from mypy.errors import (
+    ErrorInfo,
+    Errors,
+    ErrorWatcher,
+    IterationDependentErrors,
+    IterationErrorWatcher,
+    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
@@ -598,26 +605,15 @@ def accept_loop(
             # on without bound otherwise)
             widened_old = len(self.widened_vars)
 
-            # 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_errors = IterationDependentErrors()
             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()
 
-                    with LoopErrorWatcher(self.msg.errors) as watcher:
+                    with IterationErrorWatcher(self.msg.errors, iter_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)
@@ -639,29 +635,10 @@ def accept_loop(
                 if iter == 20:
                     raise RuntimeError("Too many iterations when checking a loop")
 
-            # 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)
+            for error_info in watcher.yield_error_infos():
+                self.msg.fail(*error_info[:2], code=error_info[2])
+            for note_info in watcher.yield_note_infos(self.options):
+                self.note(*note_info)
 
             # If exit_condition is set, assume it must be False on exit from the loop:
             if exit_condition:
@@ -4948,6 +4925,9 @@ def type_check_raise(self, e: Expression, s: RaiseStmt, optional: bool = False)
 
     def visit_try_stmt(self, s: TryStmt) -> None:
         """Type check a try statement."""
+
+        iter_errors = None
+
         # Our enclosing frame will get the result if the try/except falls through.
         # This one gets all possible states after the try block exited abnormally
         # (by exception, return, break, etc.)
@@ -4962,7 +4942,9 @@ def visit_try_stmt(self, s: TryStmt) -> None:
             self.visit_try_without_finally(s, try_frame=bool(s.finally_body))
             if s.finally_body:
                 # First we check finally_body is type safe on all abnormal exit paths
-                self.accept(s.finally_body)
+                iter_errors = IterationDependentErrors()
+                with IterationErrorWatcher(self.msg.errors, iter_errors) as watcher:
+                    self.accept(s.finally_body)
 
         if s.finally_body:
             # Then we try again for the more restricted set of options
@@ -4976,8 +4958,15 @@ def visit_try_stmt(self, s: TryStmt) -> None:
             # type checks in both contexts, but only the resulting types
             # from the latter context affect the type state in the code
             # that follows the try statement.)
+            assert iter_errors is not None
             if not self.binder.is_unreachable():
-                self.accept(s.finally_body)
+                with IterationErrorWatcher(self.msg.errors, iter_errors) as watcher:
+                    self.accept(s.finally_body)
+
+            for error_info in watcher.yield_error_infos():
+                self.msg.fail(*error_info[:2], code=error_info[2])
+            for note_info in watcher.yield_note_infos(self.options):
+                self.msg.note(*note_info)
 
     def visit_try_without_finally(self, s: TryStmt, try_frame: bool) -> None:
         """Type check a try statement, ignoring the finally block.

mypy/errors.py

@@ -4,13 +4,15 @@
 import sys
 import traceback
 from collections import defaultdict
-from collections.abc import Iterable
+from collections.abc import Iterable, Iterator
+from itertools import chain
 from typing import Callable, Final, NoReturn, Optional, TextIO, TypeVar
 from typing_extensions import Literal, Self, TypeAlias as _TypeAlias
 
 from mypy import errorcodes as codes
 from mypy.error_formatter import ErrorFormatter
 from mypy.errorcodes import IMPORT, IMPORT_NOT_FOUND, IMPORT_UNTYPED, ErrorCode, mypy_error_codes
+from mypy.nodes import Context
 from mypy.options import Options
 from mypy.scope import Scope
 from mypy.util import DEFAULT_SOURCE_OFFSET, is_typeshed_file
@@ -219,23 +221,43 @@ 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."""
+class IterationDependentErrors:
+    """An `IterationDependentErrors` instance serves to collect the `unreachable`,
+    `redundant-expr`, and `redundant-casts` errors, as well as the revealed types,
+    handled by the individual `IterationErrorWatcher` instances sequentially applied to
+    the same code section."""
 
-    # Meaning of the tuple items: ErrorCode, message, line, column, end_line, end_column:
-    uselessness_errors: set[tuple[ErrorCode, str, int, int, int, int]]
+    # One set of `unreachable`, `redundant-expr`, and `redundant-casts` errors per
+    # iteration step.  Meaning of the tuple items: ErrorCode, message, line, column,
+    # end_line, end_column.
+    uselessness_errors: list[set[tuple[ErrorCode, str, int, int, int, int]]]
 
-    # Meaning of the tuple items: function_or_member, line, column, end_line, end_column:
+    # One set of unreachable line numbers per iteration step.  Not only the lines where
+    # the error report occurs but really all unreachable lines.
+    unreachable_lines: list[set[int]]
+
+    # One set of revealed types for each `reveal_type` statement.  Each created set can
+    # grow during the iteration.  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) -> None:
+        self.uselessness_errors = []
+        self.unreachable_lines = []
+        self.revealed_types = defaultdict(set)
+
+
+class IterationErrorWatcher(ErrorWatcher):
+    """Error watcher that filters and separately collects `unreachable` errors,
+    `redundant-expr` and `redundant-casts` errors, and revealed types when analysing
+    code sections iteratively to help avoid making too-hasty reports."""
+
+    iteration_dependent_errors: IterationDependentErrors
 
     def __init__(
         self,
         errors: Errors,
+        iteration_dependent_errors: IterationDependentErrors,
         *,
         filter_errors: bool | Callable[[str, ErrorInfo], bool] = False,
         save_filtered_errors: bool = False,
@@ -247,31 +269,71 @@ def __init__(
             save_filtered_errors=save_filtered_errors,
             filter_deprecated=filter_deprecated,
         )
-        self.uselessness_errors = set()
-        self.unreachable_lines = set()
-        self.revealed_types = defaultdict(set)
+        self.iteration_dependent_errors = iteration_dependent_errors
+        iteration_dependent_errors.uselessness_errors.append(set())
+        iteration_dependent_errors.unreachable_lines.append(set())
 
     def on_error(self, file: str, info: ErrorInfo) -> bool:
+        """Filter out the "iteration-dependent" errors and notes and store their
+        information to handle them after iteration is completed."""
+
+        iter_errors = self.iteration_dependent_errors
 
         if info.code in (codes.UNREACHABLE, codes.REDUNDANT_EXPR, codes.REDUNDANT_CAST):
-            self.uselessness_errors.add(
+            iter_errors.uselessness_errors[-1].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))
+                iter_errors.unreachable_lines[-1].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(", "))
+                iter_errors.revealed_types[key].update(types[6:-1].split(", "))
             else:
-                self.revealed_types[key].add(types)
+                iter_errors.revealed_types[key].add(types)
             return True
 
         return super().on_error(file, info)
 
+    def yield_error_infos(self) -> Iterator[tuple[str, Context, ErrorCode]]:
+        """Report only those `unreachable`, `redundant-expr`, and `redundant-casts`
+        errors that could not be ruled out in any iteration step."""
+
+        persistent_uselessness_errors = set()
+        iter_errors = self.iteration_dependent_errors
+        for candidate in set(chain(*iter_errors.uselessness_errors)):
+            if all(
+                (candidate in errors) or (candidate[2] in lines)
+                for errors, lines in zip(
+                    iter_errors.uselessness_errors, iter_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]
+            yield error_info[1], context, error_info[0]
+
+    def yield_note_infos(self, options: Options) -> Iterator[tuple[str, Context]]:
+        """Yield all types revealed in at least one iteration step."""
+
+        for note_info, types in self.iteration_dependent_errors.revealed_types.items():
+            sorted_ = sorted(types, key=lambda typ: typ.lower())
+            if len(types) == 1:
+                revealed = sorted_[0]
+            elif options.use_or_syntax():
+                revealed = " | ".join(sorted_)
+            else:
+                revealed = 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]
+            yield f'Revealed type is "{revealed}"', context
+
 
 class Errors:
     """Container for compile errors.

test-data/unit/check-narrowing.test

@@ -2446,6 +2446,25 @@ while x is not None and b():
     x = f()
 [builtins fixtures/primitives.pyi]
 
+[case testAvoidFalseUnreachableInFinally]
+# flags: --allow-redefinition-new --local-partial-types --warn-unreachable
+def f() -> None:
+    try:
+        x = 1
+        if int():
+            x = ""
+            return
+        if int():
+            x = None
+            return
+    finally:
+        reveal_type(x)  # N: Revealed type is "Union[builtins.int, builtins.str, None]"
+        if isinstance(x, str):
+            reveal_type(x)  # N: Revealed type is "builtins.str"
+        reveal_type(x) # N: Revealed type is "Union[builtins.int, builtins.str, None]"
+
+[builtins fixtures/isinstancelist.pyi]
+
 [case testNarrowingTypeVarMultiple]
 from typing import TypeVar
 

test-data/unit/check-redefine2.test

@@ -791,8 +791,7 @@ def f3() -> None:
             x = ""
             return
     finally:
-        reveal_type(x) # N: Revealed type is "Union[builtins.int, builtins.str]" \
-            # N: Revealed type is "builtins.int"
+        reveal_type(x) # N: Revealed type is "Union[builtins.int, builtins.str]"
     reveal_type(x) # N: Revealed type is "builtins.int"
 
 def f4() -> None:

test-data/unit/check-union-error-syntax.test

@@ -55,3 +55,24 @@ from typing import Literal, Union
 x : Union[Literal[1], None]
 x = 3 # E: Incompatible types in assignment (expression has type "Literal[3]", variable has type "Optional[Literal[1]]")
 [builtins fixtures/tuple.pyi]
+
+[case testUnionSyntaxRecombined]
+# flags: --python-version 3.10 --force-union-syntax --allow-redefinition-new --local-partial-types
+# The following revealed type is recombined because the finally body is visited twice.
+try:
+    x = 1
+    x = ""
+finally:
+    reveal_type(x)  # N: Revealed type is "Union[builtins.int, builtins.str]"
+[builtins fixtures/isinstancelist.pyi]
+
+[case testOrSyntaxRecombined]
+# flags: --python-version 3.10 --no-force-union-syntax --allow-redefinition-new --local-partial-types
+# The following revealed type is recombined because the finally body is visited twice.
+# ToDo: Improve this recombination logic, especially (but not only) for the "or syntax".
+try:
+    x = 1
+    x = ""
+finally:
+    reveal_type(x)  # N: Revealed type is "builtins.int | builtins.str | builtins.str"
+[builtins fixtures/isinstancelist.pyi]