From 6656265d5ff6637f36f4c587215cae07b0930b81 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Mon, 23 Oct 2023 14:20:01 +0000
Subject: [PATCH 1/4] Add tests for issue gh-74956

---
 Lib/test/test_asyncgen.py | 117 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 117 insertions(+)

diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index a49630112af510..94e21e39424085 100644
--- a/Lib/test/test_asyncgen.py
+++ b/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 asyncio.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()

From 9ffc96de091af79874910c0f8268018ce0f1d43d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Mon, 23 Oct 2023 16:31:00 +0000
Subject: [PATCH 2/4] clean up generators in test

---
 Lib/test/test_asyncgen.py | 46 ++++++++++++++++-----------------------
 1 file changed, 19 insertions(+), 27 deletions(-)

diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index 94e21e39424085..0e5f4094804afe 100644
--- a/Lib/test/test_asyncgen.py
+++ b/Lib/test/test_asyncgen.py
@@ -1778,33 +1778,25 @@ async def run():
             async def consumer():
                 while True:
                     await asyncio.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)
+                    if (yield) is None:
+                        break
+
+            # try different combinations of asend, athrow, aclose
+            # which are clashing with an asend which is already running
+            # (and awaiting sleep(0))
+            for op, args in [("asend", ["A"]), ("athrow", [EOFError]), ("aclose", [])]:
+                agenerator = consumer()
+                await agenerator.asend(None) # start it
+                # fa will hit sleep and then fb will run
+                fa = asyncio.create_task(agenerator.asend("A"))
+                coro = getattr(agenerator, op)(*args)
+                fb = asyncio.create_task(coro)
+                await fa
+                with self.assertRaises(RuntimeError) as err:
+                    await fb
+                assert "already running" in str(err.exception)
+                with self.assertRaises(StopAsyncIteration):
+                    await agenerator.asend(None)  # close it
 
         self.loop.run_until_complete(run())
 

From 94fb0e3cb4fdd6109fad8878694e19d6217f2536 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Wed, 15 Nov 2023 09:44:35 +0000
Subject: [PATCH 3/4] Make simultaneous task test clearer

---
 Lib/test/test_asyncgen.py | 50 ++++++++++++++++++++++++++++-----------
 1 file changed, 36 insertions(+), 14 deletions(-)

diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index 0e5f4094804afe..2bd2c87d515681 100644
--- a/Lib/test/test_asyncgen.py
+++ b/Lib/test/test_asyncgen.py
@@ -1771,32 +1771,54 @@ def tearDown(self):
     def test_simultaneous_asend(self):
         """
         Verify that simultaneous use of generator by different coroutines is not
-        permitted
+        permitted.  We use Tasks to achieve this, where one task is suspended
+        in a `asyncio.sleep()` call inside the generator (during an `asend()` call),
+        and the other task attempts
+        to do an `asend()`, (or `athrow()`, or `aclose()`) on the generator.
         """
 
-        async def run():
+        
+        async def run_collision(op, *args):
+            # Two tasks are created and scheduled.  The first will sleep inside the
+            # `asend()` and the other will then attempt a second operation and fail.
+
             async def consumer():
                 while True:
+                    # task fa will sleep here, and another task will try to iterate
+                    # the generator
                     await asyncio.sleep(0)
                     if (yield) is None:
                         break
 
+            # create and start the generator
+            agenerator = consumer()
+            await agenerator.asend(None)
+
+            # start the first asend() task
+            fa = asyncio.create_task(agenerator.asend("A"))
+
+            # start the second task, which should fail (asend, athrow, aclose)
+            method = getattr(agenerator, op)
+            fb = asyncio.create_task(method(*args))
+
+            # first asend should succeed
+            await fa
+
+            # second operation should fail
+            with self.assertRaises(RuntimeError) as err:
+                await fb
+            assert "already running" in str(err.exception)
+
+            # cleanup partially run generator
+            with self.assertRaises(StopAsyncIteration):
+                await agenerator.asend(None)  # close it
+
+        async def run():
             # try different combinations of asend, athrow, aclose
             # which are clashing with an asend which is already running
             # (and awaiting sleep(0))
             for op, args in [("asend", ["A"]), ("athrow", [EOFError]), ("aclose", [])]:
-                agenerator = consumer()
-                await agenerator.asend(None) # start it
-                # fa will hit sleep and then fb will run
-                fa = asyncio.create_task(agenerator.asend("A"))
-                coro = getattr(agenerator, op)(*args)
-                fb = asyncio.create_task(coro)
-                await fa
-                with self.assertRaises(RuntimeError) as err:
-                    await fb
-                assert "already running" in str(err.exception)
-                with self.assertRaises(StopAsyncIteration):
-                    await agenerator.asend(None)  # close it
+                await run_collision(op, *args)
 
         self.loop.run_until_complete(run())
 

From 7d3b0417330295a7c3d44dfb9af2c5543cea0337 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Thu, 16 Nov 2023 09:58:13 +0000
Subject: [PATCH 4/4] lint/whitespace

---
 Lib/test/test_asyncgen.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index 2bd2c87d515681..ea2722d736f040 100644
--- a/Lib/test/test_asyncgen.py
+++ b/Lib/test/test_asyncgen.py
@@ -1777,7 +1777,6 @@ def test_simultaneous_asend(self):
         to do an `asend()`, (or `athrow()`, or `aclose()`) on the generator.
         """
 
-        
         async def run_collision(op, *args):
             # Two tasks are created and scheduled.  The first will sleep inside the
             # `asend()` and the other will then attempt a second operation and fail.