Skip to content

Commit

Permalink
Add tests which demonstrate issue #4
Browse files Browse the repository at this point in the history
Control-C and/or signalling with SIGINT doesn't result in clean
cancellation and teardown of a program that has spawned subproceses.
These tests demonstrate this issue as is documented in
#4.
  • Loading branch information
goodboy committed Feb 17, 2020
1 parent b1f366f commit b4c8750
Showing 1 changed file with 60 additions and 0 deletions.
60 changes: 60 additions & 0 deletions tests/core/test_open_in_process.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import pickle
import signal
import os

import pytest
import trio
Expand All @@ -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():
Expand Down Expand Up @@ -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)

0 comments on commit b4c8750

Please sign in to comment.