Make warn-unreachable understand exception-swallowing contextmanagers

mypy/checker.py

@@ -36,7 +36,7 @@
     UnionType, TypeVarId, TypeVarType, PartialType, DeletedType, UninhabitedType, TypeVarDef,
     true_only, false_only, function_type, is_named_instance, union_items, TypeQuery, LiteralType,
     is_optional, remove_optional, TypeTranslator, StarType, get_proper_type, ProperType,
-    get_proper_types
+    get_proper_types, is_literal_type
 )
 from mypy.sametypes import is_same_type
 from mypy.messages import (
@@ -3341,12 +3341,40 @@ def check_incompatible_property_override(self, e: Decorator) -> None:
                     self.fail(message_registry.READ_ONLY_PROPERTY_OVERRIDES_READ_WRITE, e)
 
     def visit_with_stmt(self, s: WithStmt) -> None:
+        exceptions_maybe_suppressed = False
         for expr, target in zip(s.expr, s.target):
             if s.is_async:
-                self.check_async_with_item(expr, target, s.unanalyzed_type is None)
+                exit_ret_type = self.check_async_with_item(expr, target, s.unanalyzed_type is None)
             else:
-                self.check_with_item(expr, target, s.unanalyzed_type is None)
-        self.accept(s.body)
+                exit_ret_type = self.check_with_item(expr, target, s.unanalyzed_type is None)
+
+            # Based on the return type, determine if this context manager 'swallows'
+            # exceptions or not. We determine this using a heuristic based on the
+            # return type of the __exit__ method -- see the discussion in
+            # https://github.com/python/mypy/issues/7214 and the section about context managers
+            # in https://github.com/python/typeshed/blob/master/CONTRIBUTING.md#conventions
+            # for more details.
+
+            exit_ret_type = get_proper_type(exit_ret_type)
+            if is_literal_type(exit_ret_type, "builtins.bool", False):
+                continue
+
+            if (is_literal_type(exit_ret_type, "builtins.bool", True)
+                    or (isinstance(exit_ret_type, Instance)
+                        and exit_ret_type.type.fullname() == 'builtins.bool'
+                        and state.strict_optional)):
+                # Note: if strict-optional is disabled, this bool instance
+                # could actually be an Optional[bool].
+                exceptions_maybe_suppressed = True
+
+        if exceptions_maybe_suppressed:
+            # Treat this 'with' block in the same way we'd treat a 'try: BODY; except: pass'
+            # block. This means control flow can continue after the 'with' even if the 'with'
+            # block immediately returns.
+            with self.binder.frame_context(can_skip=True, try_frame=True):
+                self.accept(s.body)
+        else:
+            self.accept(s.body)
 
     def check_untyped_after_decorator(self, typ: Type, func: FuncDef) -> None:
         if not self.options.disallow_any_decorated or self.is_stub:
@@ -3356,7 +3384,7 @@ def check_untyped_after_decorator(self, typ: Type, func: FuncDef) -> None:
             self.msg.untyped_decorated_function(typ, func)
 
     def check_async_with_item(self, expr: Expression, target: Optional[Expression],
-                              infer_lvalue_type: bool) -> None:
+                              infer_lvalue_type: bool) -> Type:
         echk = self.expr_checker
         ctx = echk.accept(expr)
         obj = echk.check_method_call_by_name('__aenter__', ctx, [], [], expr)[0]
@@ -3365,20 +3393,22 @@ def check_async_with_item(self, expr: Expression, target: Optional[Expression],
         if target:
             self.check_assignment(target, self.temp_node(obj, expr), infer_lvalue_type)
         arg = self.temp_node(AnyType(TypeOfAny.special_form), expr)
-        res = echk.check_method_call_by_name(
-            '__aexit__', ctx, [arg] * 3, [nodes.ARG_POS] * 3, expr)[0]
-        echk.check_awaitable_expr(
+        res, _ = echk.check_method_call_by_name(
+            '__aexit__', ctx, [arg] * 3, [nodes.ARG_POS] * 3, expr)
+        return echk.check_awaitable_expr(
             res, expr, message_registry.INCOMPATIBLE_TYPES_IN_ASYNC_WITH_AEXIT)
 
     def check_with_item(self, expr: Expression, target: Optional[Expression],
-                        infer_lvalue_type: bool) -> None:
+                        infer_lvalue_type: bool) -> Type:
         echk = self.expr_checker
         ctx = echk.accept(expr)
         obj = echk.check_method_call_by_name('__enter__', ctx, [], [], expr)[0]
         if target:
             self.check_assignment(target, self.temp_node(obj, expr), infer_lvalue_type)
         arg = self.temp_node(AnyType(TypeOfAny.special_form), expr)
-        echk.check_method_call_by_name('__exit__', ctx, [arg] * 3, [nodes.ARG_POS] * 3, expr)
+        res, _ = echk.check_method_call_by_name(
+            '__exit__', ctx, [arg] * 3, [nodes.ARG_POS] * 3, expr)
+        return res
 
     def visit_print_stmt(self, s: PrintStmt) -> None:
         for arg in s.args:

mypy/types.py

@@ -2406,6 +2406,17 @@ def remove_optional(typ: Type) -> ProperType:
         return typ
 
 
+def is_literal_type(typ: ProperType, fallback_fullname: str, value: LiteralValue) -> bool:
+    """Check if this type is a LiteralType with the given fallback type and value."""
+    if isinstance(typ, Instance) and typ.last_known_value:
+        typ = typ.last_known_value
+    if not isinstance(typ, LiteralType):
+        return False
+    if typ.fallback.type.fullname() != fallback_fullname:
+        return False
+    return typ.value == value
+
+
 @overload
 def get_proper_type(typ: None) -> None: ...
 @overload  # noqa