test for log workers

Author: willmcgugan

src/textual/__init__.py

@@ -105,7 +105,7 @@ def __call__(self, *args: object, **kwargs) -> None:
                     )
                     print(*print_args)
                 return
-        if app.devtools is None or not app.devtools.is_connected:
+        if not app._is_devtools_connected:
             return
 
         current_frame = inspect.currentframe()
@@ -198,4 +198,10 @@ def worker(self) -> Logger:
     from textual import log
     log(locals())
     ```
+
+!!! note
+    This logger will only work if there is an active app in the current thread.
+    Use `app.log` to write logs from a thread without an active app.
+
+
 """

src/textual/app.py

@@ -842,6 +842,11 @@ def __init__(
                     )
                 )
 
+    @property
+    def _is_devtools_connected(self) -> bool:
+        """Is the app connected to the devtools?"""
+        return self.devtools is not None and self.devtools.is_connected
+
     @cached_property
     def _exception_event(self) -> asyncio.Event:
         """An event that will be set when the first exception is encountered."""

tests/test_logger.py

@@ -0,0 +1,42 @@
+import inspect
+from typing import Any
+
+from textual import work
+from textual._log import LogGroup, LogVerbosity
+from textual.app import App
+
+
+async def test_log_from_worker() -> None:
+    """Check that log calls from threaded workers call app._log"""
+
+    log_messages: list[tuple] = []
+
+    class LogApp(App):
+
+        def _log(
+            self,
+            group: LogGroup,
+            verbosity: LogVerbosity,
+            _textual_calling_frame: inspect.Traceback,
+            *objects: Any,
+            **kwargs,
+        ) -> None:
+            log_messages.append(objects)
+
+        @property
+        def _is_devtools_connected(self):
+            """Fake connected devtools."""
+            return True
+
+        def on_mount(self) -> None:
+            self.do_work()
+            self.call_after_refresh(self.exit)
+
+        @work(thread=True)
+        def do_work(self) -> None:
+            self.log("HELLO from do_work")
+
+    app = LogApp()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        assert ("HELLO from do_work",) in log_messages