Add (so far) hanging forever tests on teardown

Author: touilleMan

tests/test_misc.py

@@ -156,3 +156,60 @@ async def cancel_sleep():
 
         await loop.run_asyncio(cancel_sleep)
         assert owch == 0
+
+
+@pytest.mark.trio
+async def test_wrong_context_manager_order():
+    async def work_in_asyncio():
+        await asyncio.sleep(0)
+
+    async def runner(*, task_status=trio.TASK_STATUS_IGNORED):
+        await trio_asyncio.run_asyncio(work_in_asyncio)
+        try:
+            task_status.started()
+            await trio.sleep_forever()
+        finally:
+            await trio_asyncio.run_asyncio(work_in_asyncio)
+
+    async with trio.open_nursery() as nursery:
+        async with trio_asyncio.open_loop():
+            await nursery.start(runner)
+
+
+@pytest.mark.trio
+async def test_keyboard_interrupt_teardown():
+    asyncio_loop_closed = trio.Event()
+
+    async def work_in_trio_no_matter_what(*, task_status=trio.TASK_STATUS_IGNORED):
+        await trio_asyncio.run_asyncio(work_in_asyncio)
+        try:
+            # KeyboardInterrupt won't cancel this coroutine thanks to the shield
+            with trio.open_cancel_scope(shield=True):
+                task_status.started()
+                await asyncio_loop_closed.wait()
+        finally:
+            # Hence this call will be exceuted after run_asyncio_loop is cancelled
+            await trio_asyncio.run_asyncio(work_in_asyncio)
+
+    async def work_in_asyncio():
+        await asyncio.sleep(0)
+
+    async def run_asyncio_loop(nursery, *, task_status=trio.TASK_STATUS_IGNORED):
+        with trio.open_cancel_scope() as cancel_scope:
+            try:
+                async with trio_asyncio.open_loop():
+                    # Starting a coroutine from here make it inherit the access
+                    # to the asyncio loop context manager
+                    await nursery.start(work_in_trio_no_matter_what)
+                    task_status.started(cancel_scope)
+                    await trio.sleep_forever()
+            finally:
+                asyncio_loop_closed.set()
+
+    import signal
+    import threading
+    with pytest.raises(KeyboardInterrupt):
+        async with trio.open_nursery() as nursery:
+            await nursery.start(run_asyncio_loop, nursery)
+            # Trigger KeyboardInterrupt that should propagate accross the coroutines
+            signal.pthread_kill(threading.get_ident(), signal.SIGINT)