Add tests which demonstrate issue #4

tests/core/test_open_in_process.py

@@ -1,5 +1,6 @@
 import pickle
 import signal
+import os
 
 import pytest
 import trio
@@ -20,6 +21,22 @@ async def do_sleep_forever():
     assert proc.returncode == 15
 
 
+@pytest.mark.trio
+async def test_open_in_proc_cancel_via_fail_after():
+    """Test that a ``trio`` initated timeout cancellation fails as expected.
+    """
+    async def do_sleep_forever():
+        import trio
+
+        await trio.sleep_forever()
+
+    with pytest.raises(trio.TooSlowError):
+        with trio.fail_after(0.5):
+            async with open_in_process(do_sleep_forever):
+                # should block forever here
+                pass
+
+
 @pytest.mark.trio
 async def test_open_in_proc_kill_while_running():
     async def do_sleep_forever():
@@ -84,3 +101,46 @@ async def sleep_forever():
             async with open_in_process(sleep_forever) as proc:
                 raise KeyboardInterrupt
         assert proc.returncode == 2
+
+
+@pytest.mark.trio
+async def test_open_in_proc_cancel_via_SIGINT():
+    """Ensure that a control-C (SIGINT) signal cancels both the parent and
+    child processes in trionic fashion
+    """
+    pid = os.getpid()
+
+    async def do_sleep_forever():
+        import trio
+
+        await trio.sleep_forever()
+
+    with trio.fail_after(2):
+        async with open_in_process(do_sleep_forever):
+            os.kill(pid, signal.SIGINT)
+            await trio.sleep_forever()
+
+
+@pytest.mark.trio
+async def test_open_in_proc_cancel_via_SIGINT_other_task():
+    """Ensure that a control-C (SIGINT) signal cancels both the parent
+    and child processes in trionic fashion even a subprocess is started
+    from a seperate ``trio`` child  task.
+    """
+    pid = os.getpid()
+
+    async def do_sleep_forever():
+        import trio
+
+        await trio.sleep_forever()
+
+    async def spawn_and_sleep_forever(task_status=trio.TASK_STATUS_IGNORED):
+        async with open_in_process(do_sleep_forever):
+            task_status.started()
+            await trio.sleep_forever()
+
+    with trio.fail_after(2):
+        # should never timeout since SIGINT should cancel the current program
+        async with trio.open_nursery() as n:
+            await n.start(spawn_and_sleep_forever)
+            os.kill(pid, signal.SIGINT)

tests/core/test_run_in_process.py

@@ -33,3 +33,27 @@ async def raise_err():
     with trio.fail_after(2):
         with pytest.raises(ValueError, match="Some err"):
             await run_in_process(raise_err)
+
+
+@pytest.mark.trio
+async def test_run_in_process_timeout_via_cancel():
+
+    async def sleep():
+        import trio
+        await trio.sleep_forever()
+
+    with pytest.raises(trio.TooSlowError):
+        with trio.fail_after(0.5):
+            await run_in_process(sleep)
+
+
+@pytest.mark.trio
+async def test_run_in_process_timeout_via_cancel_inside_child():
+
+    async def do_sleep():
+        import trio
+        with trio.fail_after(0.2):
+            await trio.sleep(float('inf'))
+
+    with pytest.raises(trio.TooSlowError):
+        await run_in_process(do_sleep)