address comments

Author: hugues-aff

mypy/checkexpr.py

@@ -8,7 +8,7 @@
 )
 from typing_extensions import ClassVar, Final, overload, TypeAlias as _TypeAlias
 
-from mypy.errors import report_internal_error
+from mypy.errors import ErrorWatcher, report_internal_error
 from mypy.typeanal import (
     has_any_from_unimported_type, check_for_explicit_any, set_any_tvars, expand_type_alias,
     make_optional_type,
@@ -2293,26 +2293,28 @@ def visit_comparison_expr(self, e: ComparisonExpr) -> Type:
                     self.msg.add_errors(local_errors)
             elif operator in operators.op_methods:
                 method = self.get_operator_method(operator)
-                err_count = self.msg.errors.total_errors()
-                sub_result, method_type = self.check_op(method, left_type, right, e,
-                                                        allow_reverse=True)
-                # Only show dangerous overlap if there are no other errors. See
-                # testCustomEqCheckStrictEquality for an example.
-                if self.msg.errors.total_errors() == err_count and operator in ('==', '!='):
-                    right_type = self.accept(right)
-                    # We suppress the error if there is a custom __eq__() method on either
-                    # side. User defined (or even standard library) classes can define this
-                    # to return True for comparisons between non-overlapping types.
-                    if (not custom_special_method(left_type, '__eq__') and
-                            not custom_special_method(right_type, '__eq__')):
-                        # Also flag non-overlapping literals in situations like:
-                        #    x: Literal['a', 'b']
-                        #    if x == 'c':
-                        #        ...
-                        left_type = try_getting_literal(left_type)
-                        right_type = try_getting_literal(right_type)
-                        if self.dangerous_comparison(left_type, right_type):
-                            self.msg.dangerous_comparison(left_type, right_type, 'equality', e)
+
+                with ErrorWatcher(self.msg.errors) as w:
+                    sub_result, method_type = self.check_op(method, left_type, right, e,
+                                                            allow_reverse=True)
+
+                    # Only show dangerous overlap if there are no other errors. See
+                    # testCustomEqCheckStrictEquality for an example.
+                    if not w.has_new_errors() and operator in ('==', '!='):
+                        right_type = self.accept(right)
+                        # We suppress the error if there is a custom __eq__() method on either
+                        # side. User defined (or even standard library) classes can define this
+                        # to return True for comparisons between non-overlapping types.
+                        if (not custom_special_method(left_type, '__eq__') and
+                                not custom_special_method(right_type, '__eq__')):
+                            # Also flag non-overlapping literals in situations like:
+                            #    x: Literal['a', 'b']
+                            #    if x == 'c':
+                            #        ...
+                            left_type = try_getting_literal(left_type)
+                            right_type = try_getting_literal(right_type)
+                            if self.dangerous_comparison(left_type, right_type):
+                                self.msg.dangerous_comparison(left_type, right_type, 'equality', e)
 
             elif operator == 'is' or operator == 'is not':
                 right_type = self.accept(right)  # validate the right operand

mypy/errors.py

@@ -4,8 +4,8 @@
 from mypy.backports import OrderedDict
 from collections import defaultdict
 
-from typing import Tuple, List, TypeVar, Set, Dict, Optional, TextIO, Callable
-from typing_extensions import Final
+from typing import Any, Tuple, List, TypeVar, Set, Dict, Optional, TextIO, Callable
+from typing_extensions import Final, Literal
 
 from mypy.scope import Scope
 from mypy.options import Options
@@ -122,6 +122,32 @@ def __init__(self,
                    Optional[ErrorCode]]
 
 
+class ErrorWatcher:
+    """Context manager that can be used to keep track of new errors recorded
+    around a given operation.
+    """
+    def __init__(self, errors: 'Errors'):
+        self.errors = errors
+        self._has_new_errors = False
+
+    def __enter__(self) -> 'ErrorWatcher':
+        self.errors._watchers.add(self)
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
+        self.errors._watchers.remove(self)
+        return False
+
+    def on_error(self, file: str, info: ErrorInfo) -> None:
+        """Handler called when a new error is recorded.
+
+        The default implementation just sets the has_new_errors flag"""
+        self._has_new_errors = True
+
+    def has_new_errors(self) -> bool:
+        return self._has_new_errors
+
+
 class Errors:
     """Container for compile errors.
 
@@ -135,9 +161,7 @@ class Errors:
     error_info_map: Dict[str, List[ErrorInfo]]
 
     # optimization for legacy codebases with many files with errors
-    has_real_errors: bool = False
-    has_blockers: bool = False
-    total_error_count: int = 0
+    has_blockers: Set[str]
 
     # Files that we have reported the errors for
     flushed_files: Set[str]
@@ -183,6 +207,8 @@ class Errors:
     # in some cases to avoid reporting huge numbers of errors.
     seen_import_error = False
 
+    _watchers: Set[ErrorWatcher] = set()
+
     def __init__(self,
                  show_error_context: bool = False,
                  show_column_numbers: bool = False,
@@ -214,6 +240,7 @@ def initialize(self) -> None:
         self.used_ignored_lines = defaultdict(lambda: defaultdict(list))
         self.ignored_files = set()
         self.only_once_messages = set()
+        self.has_blockers = set()
         self.scope = None
         self.target_module = None
         self.seen_import_error = False
@@ -239,9 +266,6 @@ def copy(self) -> 'Errors':
         new.seen_import_error = self.seen_import_error
         return new
 
-    def total_errors(self) -> int:
-        return self.total_error_count
-
     def set_ignore_prefix(self, prefix: str) -> None:
         """Set path prefix that will be removed from all paths."""
         prefix = os.path.normpath(prefix)
@@ -362,11 +386,10 @@ def _add_error_info(self, file: str, info: ErrorInfo) -> None:
         if file not in self.error_info_map:
             self.error_info_map[file] = []
         self.error_info_map[file].append(info)
-        self.total_error_count += 1
-        if info.severity == 'error':
-            self.has_real_errors = True
+        for w in self._watchers:
+            w.on_error(file, info)
         if info.blocker:
-            self.has_blockers = True
+            self.has_blockers.add(file)
         if info.code is IMPORT:
             self.seen_import_error = True
 
@@ -486,12 +509,16 @@ def clear_errors_in_targets(self, path: str, targets: Set[str]) -> None:
         """Remove errors in specific fine-grained targets within a file."""
         if path in self.error_info_map:
             new_errors = []
+            has_blocker = False
             for info in self.error_info_map[path]:
                 if info.target not in targets:
                     new_errors.append(info)
+                    has_blocker |= info.blocker
                 elif info.only_once:
                     self.only_once_messages.remove(info.message)
             self.error_info_map[path] = new_errors
+            if not has_blocker and path in self.has_blockers:
+                self.has_blockers.remove(path)
 
     def generate_unused_ignore_errors(self, file: str) -> None:
         ignored_lines = self.ignored_lines[file]
@@ -561,18 +588,14 @@ def is_errors(self) -> bool:
         """Are there any generated messages?"""
         return bool(self.error_info_map)
 
-    def is_real_errors(self) -> bool:
-        """Are there any generated errors (not just notes, for example)?"""
-        return self.has_real_errors
-
     def is_blockers(self) -> bool:
         """Are the any errors that are blockers?"""
-        return self.has_blockers
+        return bool(self.has_blockers)
 
     def blocker_module(self) -> Optional[str]:
         """Return the module with a blocking error, or None if not possible."""
-        for errs in self.error_info_map.values():
-            for err in errs:
+        for path in self.has_blockers:
+            for err in self.error_info_map[path]:
                 if err.blocker:
                     return err.module
         return None