Call from thread method

tests/test_concurrency.py

@@ -0,0 +1,50 @@
+import pytest
+
+from threading import Thread
+from textual.app import App, ComposeResult
+from textual.widgets import TextLog
+
+
+def test_call_from_thread_app_not_running():
+    app = App()
+
+    # Should fail if app is not running
+    with pytest.raises(RuntimeError):
+        app.call_from_thread(print)
+
+
+def test_call_from_thread():
+    class BackgroundThread(Thread):
+        """A background thread which will modify app in some way."""
+
+        def __init__(self, app: App) -> None:
+            self.app = app
+            super().__init__()
+
+        def run(self) -> None:
+            def write_stuff(text: str) -> None:
+                """Write stuff to a widget."""
+                self.app.query_one(TextLog).write(text)
+
+            self.app.call_from_thread(write_stuff, "Hello")
+            # Exit the app with a code we can assert
+            self.app.call_from_thread(self.app.exit, 123)
+
+    class ThreadTestApp(App):
+        """Trivial app with a single widget."""
+
+        def compose(self) -> ComposeResult:
+            yield TextLog()
+
+        def on_ready(self) -> None:
+            """Launch a thread which will modify the app."""
+            try:
+                self.call_from_thread(print)
+            except RuntimeError as error:
+                self._runtime_error = error
+            BackgroundThread(self).start()
+
+    app = ThreadTestApp()
+    result = app.run(headless=True, size=(80, 24))
+    assert isinstance(app._runtime_error, RuntimeError)
+    assert result == 123