69 lines
1.7 KiB
Python
69 lines
1.7 KiB
Python
"""
|
|
Multiple python programs invoking the runtime.
|
|
"""
|
|
import platform
|
|
import time
|
|
|
|
import pytest
|
|
import trio
|
|
import tractor
|
|
from tractor._testing import (
|
|
tractor_test,
|
|
)
|
|
from conftest import (
|
|
sig_prog,
|
|
_INT_SIGNAL,
|
|
_INT_RETURN_CODE,
|
|
)
|
|
|
|
|
|
def test_abort_on_sigint(daemon):
|
|
assert daemon.returncode is None
|
|
time.sleep(0.1)
|
|
sig_prog(daemon, _INT_SIGNAL)
|
|
assert daemon.returncode == _INT_RETURN_CODE
|
|
|
|
# XXX: oddly, couldn't get capfd.readouterr() to work here?
|
|
if platform.system() != 'Windows':
|
|
# don't check stderr on windows as its empty when sending CTRL_C_EVENT
|
|
assert "KeyboardInterrupt" in str(daemon.stderr.read())
|
|
|
|
|
|
@tractor_test
|
|
async def test_cancel_remote_arbiter(daemon, reg_addr):
|
|
assert not tractor.current_actor().is_arbiter
|
|
async with tractor.get_registry(*reg_addr) as portal:
|
|
await portal.cancel_actor()
|
|
|
|
time.sleep(0.1)
|
|
# the arbiter channel server is cancelled but not its main task
|
|
assert daemon.returncode is None
|
|
|
|
# no arbiter socket should exist
|
|
with pytest.raises(OSError):
|
|
async with tractor.get_registry(*reg_addr) as portal:
|
|
pass
|
|
|
|
|
|
def test_register_duplicate_name(daemon, reg_addr):
|
|
|
|
async def main():
|
|
|
|
async with tractor.open_nursery(
|
|
registry_addrs=[reg_addr],
|
|
) as n:
|
|
|
|
assert not tractor.current_actor().is_arbiter
|
|
|
|
p1 = await n.start_actor('doggy')
|
|
p2 = await n.start_actor('doggy')
|
|
|
|
async with tractor.wait_for_actor('doggy') as portal:
|
|
assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
|
|
|
|
await n.cancel()
|
|
|
|
# run it manually since we want to start **after**
|
|
# the other "daemon" program
|
|
trio.run(main)
|