Merge pull request #6 from A5rocks/into-the-future

Update monitor to pass basic tests

Author: A5rocks

.github/workflows/ci.yml

@@ -23,8 +23,7 @@ jobs:
           allow-prereleases: true
 
       - run: pip install uv nox
-      - run: nox -s test
-      # TODO: combine coverage
+      - run: nox -s test test_oldest
 
       - name: upload coverage
         if: always()

noxfile.py

@@ -18,6 +18,37 @@ def precommit(session):
 def test(session):
     session.install(".")
     session.install("-r", "test-requirements.txt")
+    run_tests(session)
+
+
+@nox.session(venv_backend="uv")
+def test_oldest(session):
+    pyver = session.run("python", "--version", silent=True)
+    pyver = ".".join(pyver.split(" ")[1].split(".")[:2])
+    # based on classifiers
+    possible_trio = {
+        "3.13": "trio>=0.26.0",
+        "3.12": "trio>=0.23.0",
+        "3.11": "trio>=0.21.0",
+        "3.10": "trio>=0.20.0",
+        "3.9": "trio",
+    }
+
+    session.install(".", possible_trio[pyver], "--resolution=lowest-direct")
+    session.install("pip")
+    versions = session.run("pip", "freeze", silent=True)
+
+    # make sure we don't change the trio version
+    session.install(
+        "-r",
+        "test-requirements.txt",
+        *[line for line in versions.split() if line.startswith("trio==")],
+    )
+
+    run_tests(session)
+
+
+def run_tests(session):
     session.run(
         "coverage",
         "run",

pyproject.toml

@@ -9,8 +9,8 @@ name = "trio-monitor"
 description = "A monitor utility for Trio"
 readme = "README.rst"
 license = "MIT OR Apache-2.0"
-dependencies = ["trio"]
-requires-python = ">=3.5"
+dependencies = ["trio>=0.19.0"]
+requires-python = ">=3.9"
 authors = [
 	{ name = "Lura Skye", email = "[email protected]" }
 ]
@@ -68,3 +68,7 @@ source = ["src", "**/.nox/test*/**/site-packages"]
 [tool.coverage.report]
 precision = 1
 exclude_lines = ["pragma: no cover", "abc.abstractmethod"]
+
+[tool.uv]
+# make sure that nox reinstalls trio-monitor
+reinstall-package = ["trio-monitor"]

src/trio_monitor/__init__.py

@@ -1 +1,3 @@
 """Top-level package for trio-monitor."""
+
+from .monitor import Monitor as Monitor

src/trio_monitor/_tests/test_example.py

@@ -0,0 +1,91 @@
+import trio
+import trio_monitor
+import pytest
+
+
+def run_with_monitor(async_fn, monitor):
+    trio.run(
+        async_fn,
+        instruments=[monitor],
+        clock=trio.testing.MockClock(autojump_threshold=0),
+    )
+
+
+def test_it_does_not_crash():
+    async def main():
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(trio.sleep, 1)
+
+    run_with_monitor(main, trio_monitor.Monitor())
+
+
+def test_it_can_start_a_server():
+    async def main():
+        async with trio.open_nursery() as nursery:
+            listeners = await nursery.start(trio.serve_tcp, monitor.listen_on_stream, 0)
+
+            # make sure it can handle connections
+            for listener in listeners:
+                for _ in range(2):
+                    stream = await trio.testing.open_stream_to_socket_listener(listener)
+                    async with stream:
+                        buffer = b""
+                        with trio.fail_after(1):
+                            async for part in stream:
+                                buffer += part
+                                if buffer[-6:] == b"trio> ":
+                                    break
+
+                        await stream.send_all(b"exit\r\n")
+
+            nursery.cancel_scope.cancel()
+
+    monitor = trio_monitor.Monitor()
+    run_with_monitor(main, monitor)
+
+
+@pytest.mark.xfail
+async def test_close_socket_before_start(nursery, autojump_clock):
+    monitor = trio_monitor.Monitor()
+    listeners = await nursery.start(trio.serve_tcp, monitor.listen_on_stream, 0)
+    stream = await trio.testing.open_stream_to_socket_listener(listeners[0])
+    await stream.aclose()
+    await trio.sleep(1)
+
+
+def test_monitor_program():
+    monitor = trio_monitor.Monitor()
+
+    async def main():
+        async with trio.open_nursery() as nursery:
+            tx, rx = trio.testing.memory_stream_one_way_pair()
+            nursery.start_soon(monitor.do_monitor, tx)
+
+            await trio.testing.wait_all_tasks_blocked()
+            monitor._is_monitoring = True
+
+            nursery.start_soon(trio.lowlevel.checkpoint)
+            await trio.testing.wait_all_tasks_blocked()
+
+            monitor._is_monitoring = False
+            buffer = b""
+            while True:
+                with trio.move_on_after(1) as cs:
+                    buffer += await rx.receive_some()
+
+                if cs.cancelled_caught:
+                    break
+
+            assert b"Task spawned: trio.lowlevel.checkpoint" in buffer
+            assert b"Task scheduled: trio.lowlevel.checkpoint" in buffer
+            assert b"Task stepping: trio.lowlevel.checkpoint" in buffer
+            assert b"Task finished stepping: trio.lowlevel.checkpoint" in buffer
+            assert b"Task exited: trio.lowlevel.checkpoint" in buffer
+
+            nursery.cancel_scope.cancel()
+
+        # copied from `main_loop` :/
+        monitor._is_monitoring = False
+        monitor._tx, monitor._rx = trio.open_memory_channel(100)
+
+    run_with_monitor(main, monitor)

src/trio_monitor/_tests/test_monitor.py

@@ -5,15 +5,13 @@
 import string
 import sys
 import traceback
+import importlib.metadata
 from types import FunctionType
 
-from async_generator._impl import ANextIter
-
-from trio import Queue, WouldBlock, BrokenStreamError
+from trio import WouldBlock, open_memory_channel, BrokenResourceError
 from trio._highlevel_serve_listeners import _run_handler
-from ._version import __version__
 from trio.abc import Instrument
-from trio.hazmat import current_task, Task
+from trio.lowlevel import current_task, Task
 
 
 # inspiration: https://github.com/python-trio/trio/blob/master/notes-to-self/print-task-tree.py
@@ -26,9 +24,6 @@ def walk_coro_stack(coro):
             # A real coroutine
             yield coro.cr_frame, coro.cr_frame.f_lineno
             coro = coro.cr_await
-        elif isinstance(coro, ANextIter):
-            # black hole
-            return
         else:
             # A generator decorated with @types.coroutine
             yield coro.gi_frame, coro.gi_frame.f_lineno
@@ -58,7 +53,7 @@ def __init__(self):
         self._is_monitoring = False
         # semi-arbitrary size, because otherwise we'll be dropping events
         # no clue how to make this better, alas.
-        self._monitoring_queue = Queue(capacity=100)
+        self._tx, self._rx = open_memory_channel(100)
 
     @staticmethod
     def get_root_task() -> Task:
@@ -123,9 +118,11 @@ def _add_to_monitoring_queue(self, item):
                     # if it's our own handler, skip it!
                     if loc["handler"] == self.listen_on_stream:
                         return
+            if task.coro.cr_code == self.do_monitor.__code__:
+                return
 
         try:
-            self._monitoring_queue.put_nowait(item)
+            self._tx.send_nowait(item)
         except WouldBlock:
             return
 
@@ -138,7 +135,7 @@ async def listen_on_stream(self, stream):
     async def main_loop(self, stream):
         """Runs the main loop of the monitor."""
         # send the banner
-        version = __version__
+        version = importlib.metadata.version("trio")
         await stream.send_all(
             b"Connected to the Trio monitor, using "
             b"trio " + version.encode(encoding="ascii") + b"\n"
@@ -165,7 +162,7 @@ async def main_loop(self, stream):
                 finally:
                     self._is_monitoring = False
                     # empty out the queue
-                    self._monitoring_queue = Queue(capacity=100)
+                    self._tx, self._rx = open_memory_channel(100)
 
             try:
                 fn = getattr(self, "command_{}".format(name))
@@ -195,7 +192,8 @@ async def main_loop(self, stream):
     async def do_monitor(self, stream):
         """Livefeeds information about the running program."""
         prefix = "[FEED] "
-        async for item in self._monitoring_queue:
+        while True:
+            item = await self._rx.receive()
             key = item[0]
 
             if key == "task_spawned":
@@ -232,7 +230,7 @@ async def do_monitor(self, stream):
 
             try:
                 await stream.send_all(message.encode("ascii") + b"\n")
-            except BrokenStreamError:  # client disconnected on us
+            except BrokenResourceError:  # client disconnected on us
                 return
 
     # command definitions