Add initial root actor debugger tests

examples/debugging/root_actor_breakpoint.py

@@ -0,0 +1,15 @@
+import trio
+import tractor
+
+
+async def main():
+
+    await trio.sleep(0.1)
+
+    await tractor.breakpoint()
+
+    await trio.sleep(0.1)
+
+
+if __name__ == '__main__':
+    tractor.run(main, debug_mode=True)

examples/debugging/root_actor_error.py

@@ -0,0 +1,9 @@
+import tractor
+
+
+async def main():
+    assert 0
+
+
+if __name__ == '__main__':
+    tractor.run(main, debug_mode=True)

tests/test_debugger.py

@@ -0,0 +1,106 @@
+"""
+That native debug better work!
+"""
+from os import path
+
+import pytest
+import pexpect
+
+from .test_docs_examples import repodir
+
+
+def examples_dir():
+    """Return the abspath to the examples directory.
+    """
+    return path.join(repodir(), 'examples', 'debugging/')
+
+
+def mk_cmd(ex_name: str) -> str:
+    """Generate a command suitable to pass to ``pexpect.spawn()``.
+    """
+    return ' '.join(
+        ['python',
+         path.join(examples_dir(), f'{ex_name}.py')]
+    )
+
+
+def spawn(
+    cmd: str,
+    testdir,
+) -> pexpect.spawn:
+    return testdir.spawn(
+        cmd=mk_cmd(cmd),
+        expect_timeout=3,
+    )
+
+
+@pytest.mark.parametrize(
+    'user_in_out',
+    [
+        ('c', 'AssertionError'),
+        ('q', 'AssertionError'),
+    ],
+    ids=lambda item: item[1],
+)
+def test_root_actor_error(testdir, user_in_out):
+    """Demonstrate crash handler entering pdbpp from basic error in root actor.
+    """
+    user_input, expect_err_str = user_in_out
+
+    child = spawn('root_actor_error', testdir)
+
+    # scan for the pdbpp prompt
+    child.expect("\(Pdb\+\+\)")
+
+    # make sure expected logging and error arrives
+    assert 'TTY lock acquired' in str(child.before)
+    assert 'AssertionError' in str(child.before)
+
+    # send user command
+    child.sendline(user_input)
+    child.expect('\r\n')
+    child.expect('TTY lock released')
+
+    # process should exit
+    child.expect(pexpect.EOF)
+    assert expect_err_str in str(child.before)
+
+
+@pytest.mark.parametrize(
+    'user_in_out',
+    [
+        ('c', None),
+        ('q', 'bdb.BdbQuit'),
+    ],
+    ids=lambda item: f'{item[0]} -> {item[1]}',
+)
+def test_root_actor_bp(testdir, user_in_out):
+    """Demonstrate breakpoint from in root actor.
+    """
+    user_input, expect_err_str = user_in_out
+    child = spawn('root_actor_breakpoint', testdir)
+
+    # scan for the pdbpp prompt
+    child.expect("\(Pdb\+\+\)")
+
+    assert 'Error' not in str(child.before)
+
+    # send user command
+    child.sendline(user_input)
+    child.expect('\r\n')
+
+    # process should exit
+    child.expect(pexpect.EOF)
+
+    if expect_err_str is None:
+        assert 'Error' not in str(child.before)
+    else:
+        assert expect_err_str in str(child.before)
+
+
+def test_subactor_error(testdir):
+    ...
+
+
+def test_subactor_breakpoint(testdir):
+    ...

tests/test_docs_examples.py

@@ -85,15 +85,22 @@ def run_example_in_subproc(loglevel, testdir, arb_addr):
 
 @pytest.mark.parametrize(
     'example_script',
-    [f for f in os.listdir(examples_dir()) if '__' not in f],
+    [
+        f for f in os.listdir(examples_dir())
+        if (
+            ('__' not in f) and
+            ('debugging' not in f)
+        )
+    ],
 )
 def test_example(run_example_in_subproc, example_script):
     """Load and run scripts from this repo's ``examples/`` dir as a user
     would copy and pasing them into their editor.
 
-    On windows a little more "finessing" is done to make ``multiprocessing`` play nice:
-    we copy the ``__main__.py`` into the test directory and invoke the script as a module
-    with ``python -m test_example``.
+    On windows a little more "finessing" is done to make
+    ``multiprocessing`` play nice: we copy the ``__main__.py`` into the
+    test directory and invoke the script as a module with ``python -m
+    test_example``.
     """
     ex_file = os.path.join(examples_dir(), example_script)
     with open(ex_file, 'r') as ex: