Improve the handling of "iteration dependent" errors and notes in finally clauses. (#19270)

Fixes #19269

This PR refactors the logic implemented in #19118 (which only targeted
repeatedly checked loops) and applies it to repeatedly checked finally
clauses.

I moved nearly all relevant code to the class `LoopErrorWatcher`, which
now has the more general name `IterationErrorWatcher`, to avoid code
duplication. However, one duplication is left, which concerns error
reporting. It would be nice and easy to move this functionality to
`IterationErrorWatcher`, too, but this would result in import cycles,
and I am unsure if working with `TYPE_CHECKING` and postponed importing
is acceptable in such cases (both for Mypy and Mypyc).

After the refactoring, it should not be much effort to apply the logic
to other cases where code sections are analysed iteratively. However,
the only thing that comes to my mind is the repeated checking of
functions with arguments that contain constrained type variables. I will
check it. If anyone finds a similar case and the solution is as simple
as expected, we could add the fix to this PR, of course.

Author: tyralla

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:
@@ -4960,6 +4937,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.)
@@ -4974,7 +4954,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
@@ -4988,8 +4970,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]