''' `tractor.to_actor`: one-shot single-remote-task API suite. Verifies the "spiritual successor" to (and eventual replacement of) `ActorNursery.run_in_actor()`; see https://github.com/goodboy/tractor/issues/477 ''' from functools import partial from pathlib import Path import pytest import trio import tractor from tractor import ( RemoteActorError, to_actor, ) from tractor._testing import tractor_test from tractor.msg import ptr as msgptr from tractor.msg.ptr import NamespacePath from tractor.to_actor import _api as to_actor_api async def add_one( n: int, ) -> int: return n + 1 async def raise_value_error() -> None: raise ValueError('kaboom') async def echo_control_names( value: int, /, *, name: str, portal: str, an: str, runtime_kwargs: str, ) -> dict[str, int|str]: return { 'value': value, 'name': name, 'portal': portal, 'an': an, 'runtime_kwargs': runtime_kwargs, } async def mark_task_cancellation( started_path: str, cancelled_path: str, ) -> None: Path(started_path).touch() try: await trio.sleep_forever() finally: Path(cancelled_path).touch() async def echo_startup_control( _cancel_on_startup: str, ) -> str: return _cancel_on_startup async def collect_args( *args: object, ) -> tuple[object, ...]: return args async def collect_call( *args: object, **kwargs: object, ) -> tuple[tuple[object, ...], dict[str, object]]: return args, kwargs def _non_registration_contexts( actor: tractor.Actor, ) -> dict[tuple, str]: return { key: str(ctx._nsf) for key, ctx in actor._contexts.items() if str(ctx._nsf) != ( 'tractor.discovery._registry:' 'Registrar.register_actor' ) } def test_namespace_path_retains_target_ref( monkeypatch: pytest.MonkeyPatch, ): ''' Reuse the client-side target ref when splitting its namespace path. `NamespacePath.from_ref()` previously discarded `add_one`, so `to_tuple()` imported and resolved the just-created string again. Replacing `resolve_name()` with a failure proves the retained ref supplies the tuple without a redundant lookup. The public module alias assertion also keeps internal `_api.__name__` authoritative. ''' target = NamespacePath.from_ref(add_one) def fail_resolve(name: str) -> object: raise AssertionError(f'unexpected lookup for {name!r}') monkeypatch.setattr( msgptr, 'resolve_name', fail_resolve, ) assert target.to_tuple() == ( add_one.__module__, add_one.__name__, ) assert to_actor.MODULE == to_actor_api.__name__ assert not hasattr(to_actor_api, 'MODULE') @tractor_test async def test_one_shot_in_private_nursery( start_method: str, debug_mode: bool, ): ''' No `an`/`portal` provided: a private actor-nursery is opened (and torn down) scoped to just the call. ''' assert await to_actor.run( add_one, 1, ) == 2 def test_one_shot_boots_implicit_runtime( reg_addr: tuple, start_method: str, loglevel: str, ): ''' Outside any actor-runtime `to_actor.run()` boots one implicitly (just like bare `open_nursery()` usage) configured via pass-through `runtime_kwargs`. ''' async def main() -> None: assert tractor.current_actor( err_on_no_runtime=False, ) is None result = await to_actor.run( add_one, 41, runtime_kwargs=dict( registry_addrs=[reg_addr], start_method=start_method, loglevel=loglevel, ), ) assert result == 42 trio.run(main) @tractor_test async def test_remote_error_relayed_to_caller_task( start_method: str, debug_mode: bool, ): ''' A remote task error is raised directly in the caller's task as a boxed `RemoteActorError` instead of surfacing at actor-nursery teardown as with the legacy `.run_in_actor()` API. ''' with pytest.raises(RemoteActorError) as excinfo: await to_actor.run(raise_value_error) assert excinfo.value.boxed_type is ValueError @tractor_test async def test_spawn_from_caller_nursery( start_method: str, debug_mode: bool, ): ''' Pass a caller-managed `an: ActorNursery` for the spawn. Previously `to_actor.run()` treated an actor-runtime cancel ack as process reaping, so the call returned while the child monitor and its `ActorNursery._children` record remained alive until the entire nursery exited. The assertion inside the still-open nursery proves child-process joining and record removal now complete before the one-shot call returns. ''' async with tractor.open_nursery() as an: assert await to_actor.run( add_one, 10, an=an, ) == 11 assert not an._children @tractor_test async def test_cancel_ack_failure_hard_reaps_child( monkeypatch: pytest.MonkeyPatch, start_method: str, debug_mode: bool, ): ''' Escalate a failed cancel acknowledgement and reap the child. `Portal.cancel_actor()` can return `False` when its transport is already closed without confirming runtime cancellation. The old one-shot path ignored that result, released the nursery-wide join gate and then waited forever for a still-running process. This test forces that exact result without cancelling the actor, caps the call to detect the former hang and verifies the child monitor removes its `ActorNursery._children` record before returning. ''' async def cancel_without_ack( portal: tractor.Portal, timeout: float|None = None, raise_on_timeout: bool = False, ) -> bool: assert raise_on_timeout return False monkeypatch.setattr( tractor.Portal, 'cancel_actor', cancel_without_ack, ) async with tractor.open_nursery() as an: with trio.fail_after(5): assert await to_actor.run( add_one, 20, an=an, ) == 21 assert not an._children def test_late_child_reap_registration_is_released(): ''' Preserve a nursery-wide reap request across child startup. A child monitor can checkpoint while connecting to its parent as the surrounding `ActorNursery` begins teardown. Previously the nursery signalled only already-registered child events, so a monitor registering afterward waited forever. This models that ordering by publishing the nursery-wide request first and proves the later per-child event inherits its set state immediately. ''' an = object.__new__(tractor.ActorNursery) an._join_procs = trio.Event() an._child_reap_requests = {} an._child_reaped = {} an._join_procs.set() reap_request, _ = an._register_child_reap( ('late_child', 'uid'), ) assert reap_request.is_set() @tractor_test async def test_remote_error_from_caller_nursery( start_method: str, debug_mode: bool, ): ''' With a caller-managed `an` the remote error also surfaces in the caller's task, INSIDE the nursery block, allowing inline (supervision-style) handling. ''' async with tractor.open_nursery() as an: with pytest.raises(RemoteActorError) as excinfo: await to_actor.run( raise_value_error, an=an, ) assert excinfo.value.boxed_type is ValueError @tractor_test async def test_reuse_existing_actor_via_portal( start_method: str, debug_mode: bool, ): ''' Pass `portal=` to schedule the one-shot task in an already-running actor; no spawn, no implicit reap. The low-level `Portal.run_from_ns()` assertion also proves its target kwargs remain separate from the private startup-cancel policy used by context cleanup. ''' async with tractor.open_nursery() as an: actor = tractor.current_actor() portal: tractor.Portal = await an.start_actor( 'one_shot_worker', enable_modules=[ __name__, to_actor.MODULE, ], ) contexts_before = _non_registration_contexts(actor) for i in range(3): assert await to_actor.run( add_one, i, portal=portal, ) == i + 1 assert await portal.run_from_ns( __name__, 'echo_startup_control', _cancel_on_startup='target_value', ) == 'target_value' assert _non_registration_contexts(actor) == contexts_before # still alive: caller owns the actor's lifetime. await portal.cancel_actor() @tractor_test async def test_concurrent_one_shots_from_task_nursery( start_method: str, debug_mode: bool, ): ''' The worker-pool-ish pattern from #477: concurrency is composed with a plain (caller-side) `trio` task nursery scheduling multiple one-shot calls against a shared caller-managed actor-nursery; error collection thus lives entirely in caller-code. ''' results: dict[int, int] = {} async def one_shot( an: tractor.ActorNursery, i: int, ) -> None: results[i] = await to_actor.run( add_one, i, an=an, name=f'one_shot_{i}', ) async with ( tractor.open_nursery() as an, trio.open_nursery() as tn, ): for i in range(4): tn.start_soon(one_shot, an, i) assert results == { i: i + 1 for i in range(4) } def test_rejects_sync_fn(): ''' Non-async callables error BEFORE any spawn (or even runtime-boot) happens. ''' def not_async() -> None: ... with pytest.raises(TypeError): trio.run( partial( to_actor.run, not_async, ) ) def test_rejects_streaming_fn(): ''' Async-gen (streaming) fns are not one-shot-able, same constraint as `Portal.run()`. ''' async def agen(): yield 1 with pytest.raises(TypeError): trio.run( partial( to_actor.run, agen, ) ) def test_partial_placeholder_normalization( monkeypatch: pytest.MonkeyPatch, ): ''' Preserve Python 3.14 `functools.partial` placeholder semantics. The test environment runs Python 3.13, so this installs an identity sentinel matching Python 3.14's `functools.Placeholder` API. Interleaved placeholders prove call-time positional arguments are merged in order. Undersupply and a mismatched final target signature both fail locally before actor runtime startup. ''' placeholder = object() monkeypatch.setattr( to_actor_api.functools, 'Placeholder', placeholder, raising=False, ) fn = partial( collect_args, placeholder, 2, placeholder, ) normalized_fn, args, kwargs = to_actor_api._normalize_call( fn, (1, 3, 4), ) assert normalized_fn is collect_args assert args == (1, 2, 3, 4) assert kwargs == {} with pytest.raises(TypeError, match='Not enough positional'): to_actor_api._normalize_call(fn, (1,)) with pytest.raises(TypeError, match='too many positional'): to_actor_api._normalize_call( partial(add_one, 1), (2,), ) def test_nested_partial_normalization(): ''' Flatten every retained `functools.partial` layer before RPC. CPython normally combines nested partials, but preserves the inner object when it has instance attributes. Unwrapping only the outer layer left a non-namespace-addressable partial as the RPC target. The custom attribute triggers that retained shape; the assertions prove positional ordering and outer-keyword precedence match a direct nested-partial call. ''' inner = partial( collect_call, 1, label='inner', ) inner.note = 'retain this partial layer' outer = partial( inner, 2, label='outer', ) fn, args, kwargs = to_actor_api._normalize_call( outer, (3,), ) assert fn is collect_call assert args == (1, 2, 3) assert kwargs == {'label': 'outer'} def test_rejects_portal_and_an_combo(): ''' `portal=` and `an=` are mutually exclusive placement options. ''' with pytest.raises(ValueError): trio.run( partial( to_actor.run, add_one, 1, portal=object(), an=object(), ) ) @pytest.mark.parametrize( 'placement', ['an', 'portal'], ) @pytest.mark.parametrize( 'runtime_kwargs', [ {}, {'loglevel': 'cancel'}, ], ids=['empty', 'configured'], ) def test_rejects_runtime_kwargs_with_placement( placement: str, runtime_kwargs: dict, ): ''' `runtime_kwargs` only applies when the call opens its own private actor-nursery; passing it alongside a placement opt is an error, never silently ignored. In particular, an empty dict still means the caller provided this mutually exclusive option; testing both placement modes prevents truthiness checks from accepting it before any actor runtime is started. ''' with pytest.raises(ValueError): trio.run( partial( to_actor.run, add_one, 1, **{ placement: object(), 'runtime_kwargs': runtime_kwargs, }, ) ) @tractor_test async def test_trio_style_args_and_partial_kwargs( start_method: str, debug_mode: bool, ): ''' Forward positional args and partial-bound keyword arguments. The original API captured every keyword matching an actor control, so ordinary target parameters such as `name`, `portal`, `an` and `runtime_kwargs` could not be called. This test uses a positional-only target argument plus all colliding keyword names. Binding the target keywords with `functools.partial()` proves the Trio-style calling convention keeps target inputs separate from actor controls. ''' fn = partial( echo_control_names, name='target_name', portal='target_portal', an='target_an', runtime_kwargs='target_runtime_kwargs', ) async with tractor.open_nursery() as an: result = await to_actor.run( fn, 42, an=an, name='actor_name', ) assert result == { 'value': 42, 'name': 'target_name', 'portal': 'target_portal', 'an': 'target_an', 'runtime_kwargs': 'target_runtime_kwargs', } @tractor_test async def test_portal_task_cancelled_with_local_caller( tmp_path: Path, start_method: str, debug_mode: bool, ): ''' Couple a reused portal's remote task to its local caller. The former `Portal.run()` path abandoned its remote task when the local `to_actor.run()` caller was cancelled. The target writes one file after starting and another from its cancellation `finally`. Cancelling the local task nursery and observing the second file proves `Portal.open_context()` propagated cancellation before the caller exited. A subsequent call proves the caller-owned actor was not cancelled with that task. ''' started_path = tmp_path / 'started' cancelled_path = tmp_path / 'cancelled' async with tractor.open_nursery() as an: actor = tractor.current_actor() portal: tractor.Portal = await an.start_actor( 'context_worker', enable_modules=[ __name__, to_actor.MODULE, ], ) contexts_before = _non_registration_contexts(actor) async with trio.open_nursery() as tn: tn.start_soon( partial( to_actor.run, mark_task_cancellation, str(started_path), str(cancelled_path), portal=portal, ), ) with trio.fail_after(5): while not started_path.exists(): await trio.sleep(0.01) tn.cancel_scope.cancel() assert cancelled_path.exists() assert _non_registration_contexts(actor) == contexts_before assert await to_actor.run( add_one, 1, portal=portal, ) == 2 assert _non_registration_contexts(actor) == contexts_before await portal.cancel_actor() @tractor_test async def test_context_trampoline_preserves_module_allowlist( start_method: str, debug_mode: bool, ): ''' Keep target resolution behind the actor's RPC module allowlist. Loading the target with `NamespacePath.load_ref()` would silently bypass the actor's existing module-exposure boundary. This actor exposes only the trusted trampoline, not the test module; the boxed `ModuleNotExposed` proves the trampoline delegates target resolution to `Actor._get_rpc_func()`. ''' async with tractor.open_nursery() as an: actor = tractor.current_actor() portal: tractor.Portal = await an.start_actor( 'restricted_context_worker', enable_modules=[to_actor.MODULE], ) contexts_before = _non_registration_contexts(actor) with pytest.raises(RemoteActorError) as excinfo: await to_actor.run( add_one, 1, portal=portal, ) assert excinfo.value.boxed_type is tractor.ModuleNotExposed assert _non_registration_contexts(actor) == contexts_before await portal.cancel_actor() @tractor_test async def test_portal_requires_context_trampoline( start_method: str, debug_mode: bool, ): ''' Require explicit trampoline exposure on a caller-owned actor. Automatically exposing the module in every actor weakens the RPC allowlist for actors that never use `to_actor.run()`. A portal to such an actor instead fails with the usual `ModuleNotExposed`, naming the module callers must opt into. ''' async with tractor.open_nursery() as an: actor = tractor.current_actor() portal: tractor.Portal = await an.start_actor( 'no_context_trampoline_worker', enable_modules=[__name__], ) contexts_before = _non_registration_contexts(actor) with pytest.raises(RemoteActorError) as excinfo: await to_actor.run( add_one, 1, portal=portal, ) err = excinfo.value assert err.boxed_type is tractor.ModuleNotExposed assert to_actor.MODULE in str(err) assert _non_registration_contexts(actor) == contexts_before await portal.cancel_actor()