tractor/tests/test_root_runtime.py

'''
Runtime boot/init sanity.

'''

import pytest
import trio

import tractor
from tractor._exceptions import RuntimeFailure


@tractor.context
async def open_new_root_in_sub(
    ctx: tractor.Context,
) -> None:

    async with tractor.open_root_actor():
        pass


@pytest.mark.parametrize(
    'open_root_in',
    ['root', 'sub'],
    ids='open_2nd_root_in={}'.format,
)
def test_only_one_root_actor(
    open_root_in: str,
    reg_addr: tuple,
    debug_mode: bool
):
    '''
    Verify we specially fail whenever more then one root actor
    is attempted to be opened within an already opened tree.

    '''
    async def main():
        async with tractor.open_nursery() as an:

            if open_root_in == 'root':
                async with tractor.open_root_actor(
                    registry_addrs=[reg_addr],
                ):
                    pass

            ptl: tractor.Portal = await an.start_actor(
                name='bad_rooty_boi',
                enable_modules=[__name__],
            )

            async with ptl.open_context(
                open_new_root_in_sub,
            ) as (ctx, first):
                pass

    if open_root_in == 'root':
        with pytest.raises(
            RuntimeFailure
        ) as excinfo:
            trio.run(main)

    else:
        with pytest.raises(
            tractor.RemoteActorError,
        ) as excinfo:
            trio.run(main)

        assert excinfo.value.boxed_type is RuntimeFailure


def test_implicit_root_via_first_nursery(
    reg_addr: tuple,
    debug_mode: bool
):
    '''
    The first `ActorNursery` open should implicitly call
    `_root.open_root_actor()`.

    '''
    async def main():
        async with tractor.open_nursery() as an:
            assert an._implicit_runtime_started
            assert tractor.current_actor().aid.name == 'root'

    trio.run(main)


def test_runtime_vars_unset(
    reg_addr: tuple,
    debug_mode: bool
):
    '''
    Ensure any `._state._runtime_vars` are restored to default values
    after the root actor-runtime exits!

    '''
    assert not tractor._state._runtime_vars['_debug_mode']
    async def main():
        assert not tractor._state._runtime_vars['_debug_mode']
        async with tractor.open_nursery(
            debug_mode=True,
        ):
            assert tractor._state._runtime_vars['_debug_mode']

        # after runtime closure, should be reverted!
        assert not tractor._state._runtime_vars['_debug_mode']

    trio.run(main)