Wrap `asyncio_bp.py` ex into test suite

Ensuring we can at least use `breakpoint()` from an infected actor's
`asyncio.Task` spawned via a `.to_asyncio` API.

Also includes a little `tests/devx/` reorging,
- start splitting out non-`tractor.pause()` tests into a new
  `test_pause_from_non_trio.py` for all the `.pause_from_sync()`
  use in bg-threaded or `asyncio` applications.
- factor harness commonalities to the `devx/conftest` (namely
  the `do_ctlc()` masher).
- mv `test_pause_from_sync` to the new non`-trio` mod.

NOTE, the `ctlc=True` is still failing for
`test_pause_from_asyncio_task` which is a user-happiness bug but not
anything fundamentally broken - just need to handle the `asyncio` case
in `.devx._debug.sigint_shield()`!

examples/debugging/asyncio_bp.py

@@ -1,3 +1,8 @@
+'''
+Examples of using the builtin `breakpoint()` from an `asyncio.Task`
+running in a subactor spawned with `infect_asyncio=True`.
+
+'''
 import asyncio
 
 import trio
@@ -26,15 +31,16 @@ async def bp_then_error(
     # NOTE: what happens here inside the hook needs some refinement..
     # => seems like it's still `._debug._set_trace()` but
     #    we set `Lock.local_task_in_debug = 'sync'`, we probably want
-    #    some further, at least, meta-data about the task/actoq in debug
+    #    some further, at least, meta-data about the task/actor in debug
-    #    in terms of making it clear it's asyncio mucking about.
+    #    in terms of making it clear it's `asyncio` mucking about.
     breakpoint()
 
+
     # short checkpoint / delay
-    await asyncio.sleep(0.5)
+    await asyncio.sleep(0.5)  # asyncio-side
 
     if raise_after_bp:
-        raise ValueError('blah')
+        raise ValueError('asyncio side error!')
 
     # TODO: test case with this so that it gets cancelled?
     else:
@@ -46,7 +52,7 @@ async def bp_then_error(
 @tractor.context
 async def trio_ctx(
     ctx: tractor.Context,
-    bp_before_started: bool = True,
+    bp_before_started: bool = False,
 ):
 
     # this will block until the ``asyncio`` task sends a "first"
@@ -55,7 +61,7 @@ async def trio_ctx(
 
         to_asyncio.open_channel_from(
             bp_then_error,
-            raise_after_bp=not bp_before_started,
+            # raise_after_bp=not bp_before_started,
         ) as (first, chan),
 
         trio.open_nursery() as tn,
@@ -63,9 +69,9 @@ async def trio_ctx(
         assert first == 'start'
 
         if bp_before_started:
-            await tractor.breakpoint()
+            await tractor.pause()
 
-        await ctx.started(first)
+        await ctx.started(first)  # trio-side
 
         tn.start_soon(
             to_asyncio.run_task,
@@ -77,6 +83,10 @@ async def trio_ctx(
 async def main(
     bps_all_over: bool = True,
 
+    # TODO, WHICH OF THESE HAZ BUGZ?
+    cancel_from_root: bool = False,
+    err_from_root: bool = False,
+
 ) -> None:
 
     async with tractor.open_nursery(
@@ -99,12 +109,18 @@ async def main(
 
             assert first == 'start'
 
-            if bps_all_over:
+            # pause in parent to ensure no cross-actor
-                await tractor.breakpoint()
+            # locking problems exist!
+            await tractor.pause()
 
-            # await trio.sleep_forever()
+            if cancel_from_root:
                 await ctx.cancel()
+
+            if err_from_root:
                 assert 0
+            else:
+                await trio.sleep_forever()
+
 
         # TODO: case where we cancel from trio-side while asyncio task
         # has debugger lock?

tests/devx/conftest.py

@@ -2,6 +2,7 @@
 `tractor.devx.*` tooling sub-pkg test space.
 
 '''
+import time
 from typing import (
     Callable,
 )
@@ -11,9 +12,19 @@ from pexpect.exceptions import (
     TIMEOUT,
 )
 from pexpect.spawnbase import SpawnBase
+
 from tractor._testing import (
     mk_cmd,
 )
+from tractor.devx._debug import (
+    _pause_msg as _pause_msg,
+    _crash_msg as _crash_msg,
+    _repl_fail_msg as _repl_fail_msg,
+    _ctlc_ignore_header as _ctlc_ignore_header,
+)
+from conftest import (
+    _ci_env,
+)
 
 
 @pytest.fixture
@@ -107,6 +118,9 @@ def expect(
         raise
 
 
+PROMPT = r"\(Pdb\+\)"
+
+
 def in_prompt_msg(
     child: SpawnBase,
     parts: list[str],
@@ -166,3 +180,40 @@ def assert_before(
         err_on_false=True,
         **kwargs
     )
+
+
+def do_ctlc(
+    child,
+    count: int = 3,
+    delay: float = 0.1,
+    patt: str|None = None,
+
+    # expect repl UX to reprint the prompt after every
+    # ctrl-c send.
+    # XXX: no idea but, in CI this never seems to work even on 3.10 so
+    # needs some further investigation potentially...
+    expect_prompt: bool = not _ci_env,
+
+) -> str|None:
+
+    before: str|None = None
+
+    # make sure ctl-c sends don't do anything but repeat output
+    for _ in range(count):
+        time.sleep(delay)
+        child.sendcontrol('c')
+
+        # TODO: figure out why this makes CI fail..
+        # if you run this test manually it works just fine..
+        if expect_prompt:
+            time.sleep(delay)
+            child.expect(PROMPT)
+            before = str(child.before.decode())
+            time.sleep(delay)
+
+            if patt:
+                # should see the last line on console
+                assert patt in before
+
+    # return the console content up to the final prompt
+    return before

tests/devx/test_debugger.py

@@ -21,14 +21,13 @@ from pexpect.exceptions import (
     EOF,
 )
 
-from tractor.devx._debug import (
+from .conftest import (
+    do_ctlc,
+    PROMPT,
     _pause_msg,
     _crash_msg,
     _repl_fail_msg,
 )
-from conftest import (
-    _ci_env,
-)
 from .conftest import (
     expect,
     in_prompt_msg,
@@ -70,9 +69,6 @@ has_nested_actors = pytest.mark.has_nested_actors
 # )
 
 
-PROMPT = r"\(Pdb\+\)"
-
-
 @pytest.mark.parametrize(
     'user_in_out',
     [
@@ -123,8 +119,10 @@ def test_root_actor_error(
     ids=lambda item: f'{item[0]} -> {item[1]}',
 )
 def test_root_actor_bp(spawn, user_in_out):
-    """Demonstrate breakpoint from in root actor.
+    '''
-    """
+    Demonstrate breakpoint from in root actor.
+
+    '''
     user_input, expect_err_str = user_in_out
     child = spawn('root_actor_breakpoint')
 
@@ -146,43 +144,6 @@ def test_root_actor_bp(spawn, user_in_out):
         assert expect_err_str in str(child.before)
 
 
-def do_ctlc(
-    child,
-    count: int = 3,
-    delay: float = 0.1,
-    patt: str|None = None,
-
-    # expect repl UX to reprint the prompt after every
-    # ctrl-c send.
-    # XXX: no idea but, in CI this never seems to work even on 3.10 so
-    # needs some further investigation potentially...
-    expect_prompt: bool = not _ci_env,
-
-) -> str|None:
-
-    before: str|None = None
-
-    # make sure ctl-c sends don't do anything but repeat output
-    for _ in range(count):
-        time.sleep(delay)
-        child.sendcontrol('c')
-
-        # TODO: figure out why this makes CI fail..
-        # if you run this test manually it works just fine..
-        if expect_prompt:
-            time.sleep(delay)
-            child.expect(PROMPT)
-            before = str(child.before.decode())
-            time.sleep(delay)
-
-            if patt:
-                # should see the last line on console
-                assert patt in before
-
-    # return the console content up to the final prompt
-    return before
-
-
 def test_root_actor_bp_forever(
     spawn,
     ctlc: bool,
@@ -919,138 +880,6 @@ def test_different_debug_mode_per_actor(
     )
 
 
-def test_pause_from_sync(
-    spawn,
-    ctlc: bool
-):
-    '''
-    Verify we can use the `pdbp` REPL from sync functions AND from
-    any thread spawned with `trio.to_thread.run_sync()`.
-
-    `examples/debugging/sync_bp.py`
-
-    '''
-    child = spawn('sync_bp')
-
-    # first `sync_pause()` after nurseries open
-    child.expect(PROMPT)
-    assert_before(
-        child,
-        [
-            # pre-prompt line
-            _pause_msg,
-            "<Task '__main__.main'",
-            "('root'",
-        ]
-    )
-    if ctlc:
-        do_ctlc(child)
-        # ^NOTE^ subactor not spawned yet; don't need extra delay.
-
-    child.sendline('c')
-
-    # first `await tractor.pause()` inside `p.open_context()` body
-    child.expect(PROMPT)
-
-    # XXX shouldn't see gb loaded message with PDB loglevel!
-    assert not in_prompt_msg(
-        child,
-        ['`greenback` portal opened!'],
-    )
-    # should be same root task
-    assert_before(
-        child,
-        [
-            _pause_msg,
-            "<Task '__main__.main'",
-            "('root'",
-        ]
-    )
-
-    if ctlc:
-        do_ctlc(
-            child,
-            # NOTE: setting this to 0 (or some other sufficient
-            # small val) can cause the test to fail since the
-            # `subactor` suffers a race where the root/parent
-            # sends an actor-cancel prior to it hitting its pause
-            # point; by def the value is 0.1
-            delay=0.4,
-        )
-
-    # XXX, fwiw without a brief sleep here the SIGINT might actually
-    # trigger "subactor" cancellation by its parent  before the
-    # shield-handler is engaged.
-    #
-    # => similar to the `delay` input to `do_ctlc()` below, setting
-    # this too low can cause the test to fail since the `subactor`
-    # suffers a race where the root/parent sends an actor-cancel
-    # prior to the context task hitting its pause point (and thus
-    # engaging the `sigint_shield()` handler in time); this value
-    # seems be good enuf?
-    time.sleep(0.6)
-
-    # one of the bg thread or subactor should have
-    # `Lock.acquire()`-ed
-    # (NOT both, which will result in REPL clobbering!)
-    attach_patts: dict[str, list[str]] = {
-        'subactor': [
-            "'start_n_sync_pause'",
-            "('subactor'",
-        ],
-        'inline_root_bg_thread': [
-            "<Thread(inline_root_bg_thread",
-            "('root'",
-        ],
-        'start_soon_root_bg_thread': [
-            "<Thread(start_soon_root_bg_thread",
-            "('root'",
-        ],
-    }
-    conts: int = 0  # for debugging below matching logic on failure
-    while attach_patts:
-        child.sendline('c')
-        conts += 1
-        child.expect(PROMPT)
-        before = str(child.before.decode())
-        for key in attach_patts:
-            if key in before:
-                attach_key: str = key
-                expected_patts: str = attach_patts.pop(key)
-                assert_before(
-                    child,
-                    [_pause_msg]
-                    +
-                    expected_patts
-                )
-                break
-        else:
-            pytest.fail(
-                f'No keys found?\n\n'
-                f'{attach_patts.keys()}\n\n'
-                f'{before}\n'
-            )
-
-        # ensure no other task/threads engaged a REPL
-        # at the same time as the one that was detected above.
-        for key, other_patts in attach_patts.copy().items():
-            assert not in_prompt_msg(
-                child,
-                other_patts,
-            )
-
-        if ctlc:
-            do_ctlc(
-                child,
-                patt=attach_key,
-                # NOTE same as comment above
-                delay=0.4,
-            )
-
-    child.sendline('c')
-    child.expect(EOF)
-
-
 def test_post_mortem_api(
     spawn,
     ctlc: bool,
@@ -1231,53 +1060,6 @@ def test_shield_pause(
     child.expect(EOF)
 
 
-def test_breakpoint_hook_restored(
-    spawn,
-):
-    '''
-    Ensures our actor runtime sets a custom `breakpoint()` hook
-    on open then restores the stdlib's default on close.
-
-    The hook state validation is done via `assert`s inside the
-    invoked script with only `breakpoint()` (not `tractor.pause()`)
-    calls used.
-
-    '''
-    child = spawn('restore_builtin_breakpoint')
-
-    child.expect(PROMPT)
-    assert_before(
-        child,
-        [
-            _pause_msg,
-            "<Task '__main__.main'",
-            "('root'",
-            "first bp, tractor hook set",
-        ]
-    )
-    child.sendline('c')
-    child.expect(PROMPT)
-    assert_before(
-        child,
-        [
-            "last bp, stdlib hook restored",
-        ]
-    )
-
-    # since the stdlib hook was already restored there should be NO
-    # `tractor` `log.pdb()` content from console!
-    assert not in_prompt_msg(
-        child,
-        [
-            _pause_msg,
-            "<Task '__main__.main'",
-            "('root'",
-        ],
-    )
-    child.sendline('c')
-    child.expect(EOF)
-
-
 # TODO: better error for "non-ideal" usage from the root actor.
 # -[ ] if called from an async scope emit a message that suggests
 #    using `await tractor.pause()` instead since it's less overhead
@@ -1295,7 +1077,6 @@ def test_sync_pause_from_bg_task_in_root_actor_():
     '''
     ...
 
-
 # TODO: needs ANSI code stripping tho, see `assert_before()` # above!
 def test_correct_frames_below_hidden():
     '''
@@ -1312,9 +1093,8 @@ def test_cant_pause_from_paused_task():
     '''
     Pausing from with an already paused task should raise an error.
 
-    Normally this should only happen in practise while debugging the
+    Normally this should only happen in practise while debugging the call stack of `tractor.pause()` itself, likely
-    call stack of `tractor.pause()` itself, likely by a `.pause()`
+    by a `.pause()` line somewhere inside our runtime.
-    line somewhere inside our runtime.
 
     '''
     ...

tests/devx/test_pause_from_non_trio.py

@@ -0,0 +1,329 @@
+'''
+That "foreign loop/thread" debug REPL support better ALSO WORK!
+
+Same as `test_native_pause.py`.
+All these tests can be understood (somewhat) by running the
+equivalent `examples/debugging/` scripts manually.
+
+'''
+# from functools import partial
+# import itertools
+import time
+# from typing import (
+#     Iterator,
+# )
+
+import pytest
+from pexpect.exceptions import (
+    # TIMEOUT,
+    EOF,
+)
+
+from .conftest import (
+    # _ci_env,
+    do_ctlc,
+    PROMPT,
+    # expect,
+    in_prompt_msg,
+    assert_before,
+    _pause_msg,
+    _crash_msg,
+    _ctlc_ignore_header,
+    # _repl_fail_msg,
+)
+
+
+def test_pause_from_sync(
+    spawn,
+    ctlc: bool,
+):
+    '''
+    Verify we can use the `pdbp` REPL from sync functions AND from
+    any thread spawned with `trio.to_thread.run_sync()`.
+
+    `examples/debugging/sync_bp.py`
+
+    '''
+    child = spawn('sync_bp')
+
+    # first `sync_pause()` after nurseries open
+    child.expect(PROMPT)
+    assert_before(
+        child,
+        [
+            # pre-prompt line
+            _pause_msg,
+            "<Task '__main__.main'",
+            "('root'",
+        ]
+    )
+    if ctlc:
+        do_ctlc(child)
+        # ^NOTE^ subactor not spawned yet; don't need extra delay.
+
+    child.sendline('c')
+
+    # first `await tractor.pause()` inside `p.open_context()` body
+    child.expect(PROMPT)
+
+    # XXX shouldn't see gb loaded message with PDB loglevel!
+    assert not in_prompt_msg(
+        child,
+        ['`greenback` portal opened!'],
+    )
+    # should be same root task
+    assert_before(
+        child,
+        [
+            _pause_msg,
+            "<Task '__main__.main'",
+            "('root'",
+        ]
+    )
+
+    if ctlc:
+        do_ctlc(
+            child,
+            # NOTE: setting this to 0 (or some other sufficient
+            # small val) can cause the test to fail since the
+            # `subactor` suffers a race where the root/parent
+            # sends an actor-cancel prior to it hitting its pause
+            # point; by def the value is 0.1
+            delay=0.4,
+        )
+
+    # XXX, fwiw without a brief sleep here the SIGINT might actually
+    # trigger "subactor" cancellation by its parent  before the
+    # shield-handler is engaged.
+    #
+    # => similar to the `delay` input to `do_ctlc()` below, setting
+    # this too low can cause the test to fail since the `subactor`
+    # suffers a race where the root/parent sends an actor-cancel
+    # prior to the context task hitting its pause point (and thus
+    # engaging the `sigint_shield()` handler in time); this value
+    # seems be good enuf?
+    time.sleep(0.6)
+
+    # one of the bg thread or subactor should have
+    # `Lock.acquire()`-ed
+    # (NOT both, which will result in REPL clobbering!)
+    attach_patts: dict[str, list[str]] = {
+        'subactor': [
+            "'start_n_sync_pause'",
+            "('subactor'",
+        ],
+        'inline_root_bg_thread': [
+            "<Thread(inline_root_bg_thread",
+            "('root'",
+        ],
+        'start_soon_root_bg_thread': [
+            "<Thread(start_soon_root_bg_thread",
+            "('root'",
+        ],
+    }
+    conts: int = 0  # for debugging below matching logic on failure
+    while attach_patts:
+        child.sendline('c')
+        conts += 1
+        child.expect(PROMPT)
+        before = str(child.before.decode())
+        for key in attach_patts:
+            if key in before:
+                attach_key: str = key
+                expected_patts: str = attach_patts.pop(key)
+                assert_before(
+                    child,
+                    [_pause_msg]
+                    +
+                    expected_patts
+                )
+                break
+        else:
+            pytest.fail(
+                f'No keys found?\n\n'
+                f'{attach_patts.keys()}\n\n'
+                f'{before}\n'
+            )
+
+        # ensure no other task/threads engaged a REPL
+        # at the same time as the one that was detected above.
+        for key, other_patts in attach_patts.copy().items():
+            assert not in_prompt_msg(
+                child,
+                other_patts,
+            )
+
+        if ctlc:
+            do_ctlc(
+                child,
+                patt=attach_key,
+                # NOTE same as comment above
+                delay=0.4,
+            )
+
+    child.sendline('c')
+    child.expect(EOF)
+
+
+def expect_any_of(
+    attach_patts: dict[str, list[str]],
+    child,   # what type?
+    ctlc: bool = False,
+    prompt: str = _ctlc_ignore_header,
+    ctlc_delay: float = .4,
+
+) -> list[str]:
+    '''
+    Receive any of a `list[str]` of patterns provided in
+    `attach_patts`.
+
+    Used to test racing prompts from multiple actors and/or
+    tasks using a common root process' `pdbp` REPL.
+
+    '''
+    assert attach_patts
+
+    child.expect(PROMPT)
+    before = str(child.before.decode())
+
+    for attach_key in attach_patts:
+        if attach_key in before:
+            expected_patts: str = attach_patts.pop(attach_key)
+            assert_before(
+                child,
+                expected_patts
+            )
+            break  # from for
+    else:
+        pytest.fail(
+            f'No keys found?\n\n'
+            f'{attach_patts.keys()}\n\n'
+            f'{before}\n'
+        )
+
+    # ensure no other task/threads engaged a REPL
+    # at the same time as the one that was detected above.
+    for key, other_patts in attach_patts.copy().items():
+        assert not in_prompt_msg(
+            child,
+            other_patts,
+        )
+
+    if ctlc:
+        do_ctlc(
+            child,
+            patt=prompt,
+            # NOTE same as comment above
+            delay=ctlc_delay,
+        )
+
+    return expected_patts
+    # yield child
+
+
+def test_pause_from_asyncio_task(
+    spawn,
+    ctlc: bool
+    # ^TODO, fix for `asyncio`!!
+):
+    '''
+    Verify we can use the `pdbp` REPL from an `asyncio.Task` spawned using
+    APIs in `.to_asyncio`.
+
+    `examples/debugging/asycio_bp.py`
+
+    '''
+    child = spawn('asyncio_bp')
+
+    # RACE on whether trio/asyncio task bps first
+    attach_patts: dict[str, list[str]] = {
+
+        # first pause in guest-mode (aka "infecting")
+        # `trio.Task`.
+        'trio-side': [
+            _pause_msg,
+            "<Task 'trio_ctx'",
+            "('aio_daemon'",
+        ],
+
+        # `breakpoint()` from `asyncio.Task`.
+        'asyncio-side': [
+            _pause_msg,
+            "<Task pending name='Task-2' coro=<greenback_shim()",
+            "('aio_daemon'",
+        ],
+    }
+
+    while attach_patts:
+        expect_any_of(
+            attach_patts=attach_patts,
+            child=child,
+            ctlc=ctlc,
+        )
+        child.sendline('c')
+
+    # NOW in race order,
+    # - the asyncio-task will error
+    # - the root-actor parent task will pause
+    #
+    attach_patts: dict[str, list[str]] = {
+
+        # error raised in `asyncio.Task`
+        "raise ValueError('asyncio side error!')": [
+            _crash_msg,
+            'return await chan.receive()',  # `.to_asyncio` impl internals in tb
+            "<Task 'trio_ctx'",
+            "@ ('aio_daemon'",
+            "ValueError: asyncio side error!",
+        ],
+
+        # parent-side propagation via actor-nursery/portal
+        # "tractor._exceptions.RemoteActorError: remote task raised a 'ValueError'": [
+        "remote task raised a 'ValueError'": [
+            _crash_msg,
+            "src_uid=('aio_daemon'",
+            "('aio_daemon'",
+        ],
+
+        # a final pause in root-actor
+        "<Task '__main__.main'": [
+            _pause_msg,
+            "<Task '__main__.main'",
+            "('root'",
+        ],
+    }
+    while attach_patts:
+        expect_any_of(
+            attach_patts=attach_patts,
+            child=child,
+            ctlc=ctlc,
+        )
+        child.sendline('c')
+
+    assert not attach_patts
+
+    # final boxed error propagates to root
+    assert_before(
+        child,
+        [
+            _crash_msg,
+            "<Task '__main__.main'",
+            "('root'",
+            "remote task raised a 'ValueError'",
+            "ValueError: asyncio side error!",
+        ]
+    )
+
+    if ctlc:
+        do_ctlc(
+            child,
+            # NOTE: setting this to 0 (or some other sufficient
+            # small val) can cause the test to fail since the
+            # `subactor` suffers a race where the root/parent
+            # sends an actor-cancel prior to it hitting its pause
+            # point; by def the value is 0.1
+            delay=0.4,
+        )
+
+    child.sendline('c')
+    child.expect(EOF)