subscribe: improve coverage

Also fix an issue revealed in added test.

Replicates graphql/graphql-js@4d0601b

Author: Cito

src/graphql/subscription/map_async_iterator.py

@@ -70,7 +70,7 @@ async def athrow(self, type_, value=None, traceback=None):
             if athrow:
                 await athrow(type_, value, traceback)
             else:
-                self.is_closed = True
+                await self.aclose()
                 if value is None:
                     if traceback is None:
                         raise type_

tests/subscription/test_subscribe.py

@@ -593,6 +593,160 @@ async def produces_a_payload_per_subscription_event():
         with raises(StopAsyncIteration):
             assert await anext(subscription)
 
+    @mark.asyncio
+    async def produces_a_payload_when_there_are_multiple_events():
+        pubsub = EventEmitter()
+        send_important_email, subscription = await create_subscription(pubsub)
+        payload = anext(subscription)
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright",
+                    "message": "Tests are good",
+                    "unread": True,
+                }
+            )
+            is True
+        )
+
+        assert await payload == (
+            {
+                "importantEmail": {
+                    "email": {"from": "yuzhi@graphql.org", "subject": "Alright"},
+                    "inbox": {"unread": 1, "total": 2},
+                }
+            },
+            None,
+        )
+
+        payload = anext(subscription)
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright 2",
+                    "message": "Tests are good 2",
+                    "unread": True,
+                }
+            )
+            is True
+        )
+
+        assert await payload == (
+            {
+                "importantEmail": {
+                    "email": {"from": "yuzhi@graphql.org", "subject": "Alright 2"},
+                    "inbox": {"unread": 2, "total": 3},
+                }
+            },
+            None,
+        )
+
+    @mark.asyncio
+    async def should_not_trigger_when_subscription_is_already_done():
+        pubsub = EventEmitter()
+        send_important_email, subscription = await create_subscription(pubsub)
+        payload = anext(subscription)
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright",
+                    "message": "Tests are good",
+                    "unread": True,
+                }
+            )
+            is True
+        )
+
+        assert await payload == (
+            {
+                "importantEmail": {
+                    "email": {"from": "yuzhi@graphql.org", "subject": "Alright"},
+                    "inbox": {"unread": 1, "total": 2},
+                }
+            },
+            None,
+        )
+
+        payload = anext(subscription)
+        await subscription.aclose()
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright 2",
+                    "message": "Tests are good 2",
+                    "unread": True,
+                }
+            )
+            is False
+        )
+
+        with raises(StopAsyncIteration):
+            await payload
+
+    @mark.asyncio
+    async def should_not_trigger_when_subscription_is_thrown():
+        pubsub = EventEmitter()
+        send_important_email, subscription = await create_subscription(pubsub)
+        payload = anext(subscription)
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright",
+                    "message": "Tests are good",
+                    "unread": True,
+                }
+            )
+            is True
+        )
+
+        assert await payload == (
+            {
+                "importantEmail": {
+                    "email": {"from": "yuzhi@graphql.org", "subject": "Alright"},
+                    "inbox": {"unread": 1, "total": 2},
+                }
+            },
+            None,
+        )
+
+        payload = anext(subscription)
+
+        # Throw error
+        with raises(RuntimeError) as exc_info:
+            await subscription.athrow(RuntimeError("ouch"))
+        assert str(exc_info.value) == "ouch"
+
+        # A new email arrives!
+        assert (
+            send_important_email(
+                {
+                    "from": "yuzhi@graphql.org",
+                    "subject": "Alright 2",
+                    "message": "Tests are good 2",
+                    "unread": True,
+                }
+            )
+            is False
+        )
+
+        with raises(StopAsyncIteration):
+            await payload
+
     @mark.asyncio
     async def event_order_is_correct_for_multiple_publishes():
         pubsub = EventEmitter()