Add a pytype error unused-coroutine.

PiperOrigin-RevId: 765456902

Author: pytype authors

pytype/errors/errors.py

@@ -1365,6 +1365,14 @@ def typed_dict_error(self, stack, obj, name):
       )
     self.error(stack, err_msg)
 
+  @_error_name("unused-coroutine")
+  def unused_coroutine(self, stack):
+    self.error(
+        stack,
+        "Coroutine result was not used. Did you forget to await it or assign it"
+        " to a variable?",
+    )
+
   @_error_name("final-error")
   def _overriding_final(self, stack, cls, base, name, *, is_method, details):
     desc = "method" if is_method else "class attribute"

pytype/tests/test_coroutine.py

@@ -500,9 +500,9 @@ async def func2(x: Coroutine[Any, Any, str]):
             var.func()
           return res
 
-        func1(foo.f1())
-        func1(foo.f2())
-        func2(foo.f1())
+        coro1 = func1(foo.f1())
+        coro2 = func1(foo.f2())
+        coro3 = func2(foo.f1())
       """,
           pythonpath=[d.path],
       )
@@ -512,6 +512,10 @@ async def func2(x: Coroutine[Any, Any, str]):
         import foo
         from typing import Any, Awaitable, Coroutine, List
 
+        coro1: Coroutine[Any, Any, list[str]]
+        coro2: Coroutine[Any, Any, list[str]]
+        coro3: Coroutine[Any, Any, list[str]]
+
         def func1(x: Awaitable[str]) -> Coroutine[Any, Any, List[str]]: ...
         def func2(x: Coroutine[Any, Any, str]) -> Coroutine[Any, Any, List[str]]: ...
       """,
@@ -570,7 +574,7 @@ async def worker(queue):
 
       async def main():
         queue = asyncio.Queue()
-        worker(queue)
+        await worker(queue)
     """)
     self.assertTypesMatchPytd(
         ty,
@@ -646,5 +650,58 @@ async def call_with_retry(
       """)
 
 
+class UnusedCoroutineTest(test_base.BaseTest):
+  """Tests for the unused-coroutine error."""
+
+  def test_trigger_unused_coroutine(self):
+    self.CheckWithErrors("""
+      async def sample_coro():
+        return 1
+      sample_coro()  # unused-coroutine
+    """)
+
+  def test_correct_coroutine_usage(self):
+    self.Check("""
+      import asyncio
+
+      async def sample_coro():
+        return 1
+
+      async def main():
+        await sample_coro()
+        x = sample_coro()
+        asyncio.create_task(sample_coro())
+        my_list = [sample_coro()]
+        my_dict = {"a": sample_coro()}
+        return x, my_list, my_dict
+
+      def regular_func(): return 1
+      regular_func()
+    """)
+
+  def test_wrapper_returns_unused_coroutine(self):
+    self.CheckWithErrors("""
+      async def sample_coro():
+        return 1
+      def wrapper():
+        return sample_coro()
+      wrapper()  # unused-coroutine
+    """)
+
+  def test_assign_to_underscore(self):
+    self.Check("""
+      async def sample_coro():
+        return 1
+      _ = sample_coro()
+    """)
+
+  def test_coroutine_in_used_expression(self):
+    self.CheckWithErrors("""
+      async def sample_coro():
+        return 1
+      y = 1 + sample_coro() # unsupported-operands
+    """)
+
+
 if __name__ == "__main__":
   test_base.main()

pytype/tests/test_stdlib2.py

@@ -453,14 +453,17 @@ async def iterate(x):
           pass
         else:
           pass
-      iterate(AsyncIterable())
+      iterate_coro = iterate(AsyncIterable())
     """)
     self.assertTypesMatchPytd(
         ty,
         """
       import asyncio
       from typing import Any, Coroutine, TypeVar
       _TAsyncIterable = TypeVar('_TAsyncIterable', bound=AsyncIterable)
+      
+      iterate_coro: Coroutine[Any, Any, None]
+      
       class AsyncIterable:
           def __aiter__(self: _TAsyncIterable) -> _TAsyncIterable: ...
           def __anext__(self) -> Coroutine[Any, Any, int]: ...

pytype/vm.py

@@ -1638,6 +1638,18 @@ def byte_SETUP_EXCEPT_311(self, state, op):
     return self._setup_except(state, op)
 
   def byte_POP_TOP(self, state, op):
+    """Pops and discards, checking for an unused coroutine."""
+    value_on_stack = None
+    try:
+      value_on_stack = state.top()
+    except IndexError:
+      pass
+    if value_on_stack and state.node:
+      if any(
+          isinstance(b.data, abstract.Coroutine)
+          for b in value_on_stack.Bindings(state.node)
+      ):
+        self.ctx.errorlog.unused_coroutine(self.frames)
     return state.pop_and_discard()
 
   def byte_DUP_TOP(self, state, op):