Add tests for issue gh-74956

Author: kristjanvalur

Lib/test/test_asyncgen.py

@@ -1695,6 +1695,32 @@ async def run():
 
         self.loop.run_until_complete(run())
 
+    def test_send_exhausted(self):
+        """
+        Test the behaviour of an exhausted async generator
+        """
+
+        async def run():
+            # test exhausted generator
+            async def gen():
+                yield 1
+
+            g = gen()
+            [f async for f in g]
+
+            # an exhausted generator behaves like this:
+
+            with self.assertRaises(StopAsyncIteration):
+                r = await g.asend(None)
+            with self.assertRaises(StopAsyncIteration):
+                r = await g.__anext__()
+
+            # The following behaviour may be undefined:
+            r = await g.athrow(EOFError)
+            assert r is None
+
+        self.loop.run_until_complete(run())
+
 
 class TestUnawaitedWarnings(unittest.TestCase):
     def test_asend(self):
@@ -1728,6 +1754,97 @@ async def gen():
             gc_collect()
 
 
+class TestIssueGH74956(unittest.TestCase):
+    # simultanous use of generator by different coroutines is not
+    # allowed.
+    # https://github.com/python/cpython/issues/74956
+
+    def setUp(self):
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(None)
+
+    def tearDown(self):
+        self.loop.close()
+        self.loop = None
+        asyncio.set_event_loop_policy(None)
+
+    def test_simultaneous_asend(self):
+        """
+        Verify that simultaneous use of generator by different coroutines is not
+        permitted
+        """
+
+        async def run():
+            async def consumer():
+                while True:
+                    await sleep(0)
+                    yield
+
+            agenerator = consumer()
+            await agenerator.asend(None)
+            fa = asyncio.create_task(agenerator.asend("A"))
+            fb = asyncio.create_task(agenerator.asend("B"))
+            await fa
+            with self.assertRaises(RuntimeError) as err:
+                await fb
+            assert "already running" in str(err.exception)
+
+            agenerator = consumer()
+            await agenerator.asend(None)
+            fa = asyncio.create_task(agenerator.asend("A"))
+            fb = asyncio.create_task(agenerator.athrow(EOFError))
+            await fa
+            with self.assertRaises(RuntimeError) as err:
+                await fb
+            assert "already running" in str(err.exception)
+
+            await agenerator.asend(None)
+            fa = asyncio.create_task(agenerator.asend("A"))
+            fb = asyncio.create_task(agenerator.aclose())
+            await fa
+            with self.assertRaises(RuntimeError) as err:
+                await fb
+            assert "already running" in str(err.exception)
+
+        self.loop.run_until_complete(run())
+
+    def test_ag_running(self):
+        """
+        Verify that as_running transitions correctly in
+        an async generator
+        """
+        state = 0
+
+        async def agen():
+            nonlocal state
+            state = 1
+            await asyncio.sleep(0)
+            state = 2
+            value = yield "foo"
+            state = value
+
+        a = agen()
+        assert a.ag_running is False
+        # start it running
+        coro = a.asend(None)
+        assert state == 0
+        coro.send(None)
+        assert state == 1
+        assert a.ag_running is True
+
+        # wake it from sleep and have it yield
+        with self.assertRaises(StopIteration) as v:
+            coro.send(None)
+        assert v.exception.value == "foo"
+        assert state == 2
+        assert a.ag_running is False
+
+        # finish it
+        coro = a.asend("bar")
+        self.assertRaises(StopAsyncIteration, coro.send, None)
+        assert a.ag_running is False
+        assert state == "bar"
+
 
 if __name__ == "__main__":
     unittest.main()