Robust Runner

cobald_tests/daemon/test_service.py

@@ -5,6 +5,8 @@
 import asyncio
 import contextlib
 import logging
+import signal
+import os
 
 import pytest
 
@@ -25,6 +27,7 @@ def accept(payload: ServiceRunner, name=None):
     )
     thread.start()
     if not payload.running.wait(1):
+        payload.shutdown()
         raise RuntimeError("%s failed to start" % payload)
     try:
         yield
@@ -33,24 +36,30 @@ def accept(payload: ServiceRunner, name=None):
         thread.join()
 
 
-class TestServiceRunner(object):
-    def test_no_tainting(self):
-        """Assert that no payloads may be scheduled before starting"""
+def sync_raise(what):
+    logging.info(f"raising {what}")
+    raise what
 
-        def payload():
-            return
 
-        runner = ServiceRunner()
-        runner._meta_runner.register_payload(payload, flavour=threading)
-        with pytest.raises(RuntimeError):
-            runner.accept()
+async def async_raise(what):
+    sync_raise(what)
+
+
+def sync_raise_signal(what):
+    logging.info(f"signal {what}")
+    os.kill(os.getpid(), what)
+
+
+async def async_raise_signal(what):
+    sync_raise_signal(what)
+
 
+class TestServiceRunner(object):
     def test_unique_reaper(self):
         """Assert that no two runners may fetch services"""
         with accept(ServiceRunner(accept_delay=0.1), name="outer"):
             with pytest.raises(RuntimeError):
-                with accept(ServiceRunner(accept_delay=0.1), name="inner"):
-                    pass
+                ServiceRunner(accept_delay=0.1).accept()
 
     def test_service(self):
         """Test running service classes automatically"""
@@ -133,3 +142,36 @@ async def co_pingpong(what=default):
                     break
             else:
                 assert len(reply_store) == 9
+
+    @pytest.mark.parametrize(
+        "flavour, do_sleep, do_raise",
+        (
+            (asyncio, asyncio.sleep, async_raise),
+            (trio, trio.sleep, async_raise),
+            (threading, time.sleep, sync_raise),
+        ),
+    )
+    def test_error_reporting(self, flavour, do_sleep, do_raise):
+        """Test that fatal errors do not pass silently"""
+        # errors should fail the entire runtime
+        runner = ServiceRunner(accept_delay=0.1)
+        runner.adopt(do_sleep, 5, flavour=flavour)
+        runner.adopt(do_raise, LookupError, flavour=flavour)
+        with pytest.raises(RuntimeError):
+            runner.accept()
+
+    @pytest.mark.parametrize(
+        "flavour, do_sleep, do_raise",
+        (
+            (asyncio, asyncio.sleep, async_raise_signal),
+            (trio, trio.sleep, async_raise_signal),
+            (threading, time.sleep, sync_raise_signal),
+        ),
+    )
+    def test_interrupt(self, flavour, do_sleep, do_raise):
+        """Test that KeyboardInterrupt/^C is graceful shutdown"""
+        runner = ServiceRunner(accept_delay=0.1)
+        runner.adopt(do_sleep, 5, flavour=flavour)
+        # signal.SIGINT == KeyboardInterrupt
+        runner.adopt(do_raise, signal.SIGINT, flavour=flavour)
+        runner.accept()

cobald_tests/utility/concurrent/test_meta_runner.py

@@ -2,6 +2,7 @@
 import pytest
 import time
 import asyncio
+import contextlib
 
 import trio
 
@@ -13,36 +14,22 @@ class TerminateRunner(Exception):
     pass
 
 
-def run_in_thread(payload, name, daemon=True):
-    thread = threading.Thread(target=payload, name=name, daemon=daemon)
+@contextlib.contextmanager
+def threaded_run(name=None):
+    runner = MetaRunner()
+    thread = threading.Thread(target=runner.run, name=name or str(runner), daemon=True)
     thread.start()
-    time.sleep(0.0)
+    if not runner.running.wait(1):
+        runner.stop()
+        raise RuntimeError("%s failed to start" % runner)
+    try:
+        yield runner
+    finally:
+        runner.stop()
+        thread.join()
 
 
 class TestMetaRunner(object):
-    def test_bool_payloads(self):
-        def subroutine():
-            time.sleep(0.5)
-
-        async def a_coroutine():
-            await asyncio.sleep(0.5)
-
-        async def t_coroutine():
-            await trio.sleep(0.5)
-
-        for flavour, payload in (
-            (threading, subroutine),
-            (asyncio, a_coroutine),
-            (trio, t_coroutine),
-        ):
-            runner = MetaRunner()
-            assert not bool(runner)
-            runner.register_payload(payload, flavour=flavour)
-            assert bool(runner)
-            run_in_thread(runner.run, name="test_bool_payloads %s" % flavour)
-            assert bool(runner)
-            runner.stop()
-
     @pytest.mark.parametrize("flavour", (threading,))
     def test_run_subroutine(self, flavour):
         """Test executing a subroutine"""
@@ -53,11 +40,11 @@ def with_return():
         def with_raise():
             raise KeyError("expected exception")
 
-        runner = MetaRunner()
-        result = runner.run_payload(with_return, flavour=flavour)
-        assert result == with_return()
-        with pytest.raises(KeyError):
-            runner.run_payload(with_raise, flavour=flavour)
+        with threaded_run("test_run_subroutine") as runner:
+            result = runner.run_payload(with_return, flavour=flavour)
+            assert result == with_return()
+            with pytest.raises(KeyError):
+                runner.run_payload(with_raise, flavour=flavour)
 
     @pytest.mark.parametrize("flavour", (asyncio, trio))
     def test_run_coroutine(self, flavour):
@@ -69,13 +56,11 @@ async def with_return():
         async def with_raise():
             raise KeyError("expected exception")
 
-        runner = MetaRunner()
-        run_in_thread(runner.run, name="test_run_coroutine %s" % flavour)
-        result = runner.run_payload(with_return, flavour=flavour)
-        assert result == trio.run(with_return)
-        with pytest.raises(KeyError):
-            runner.run_payload(with_raise, flavour=flavour)
-        runner.stop()
+        with threaded_run("test_run_coroutine") as runner:
+            result = runner.run_payload(with_return, flavour=flavour)
+            assert result == trio.run(with_return)
+            with pytest.raises(KeyError):
+                runner.run_payload(with_raise, flavour=flavour)
 
     @pytest.mark.parametrize("flavour", (threading,))
     def test_return_subroutine(self, flavour):
@@ -151,7 +136,6 @@ async def loop():
                 await flavour.sleep(0)
 
         runner = MetaRunner()
-
         runner.register_payload(noop, loop, flavour=flavour)
         runner.register_payload(abort, flavour=flavour)
         with pytest.raises(RuntimeError) as exc:

src/cobald/daemon/runners/_compat.py

@@ -0,0 +1,51 @@
+import sys
+import asyncio
+import inspect
+
+
+if sys.version_info >= (3, 7):
+    asyncio_run = asyncio.run
+else:
+    # almost literal backport of asyncio.run
+    def asyncio_run(main, *, debug=None):
+        assert inspect.iscoroutine(main)
+        loop = asyncio.new_event_loop()
+        try:
+            asyncio.set_event_loop(loop)
+            if debug is not None:
+                loop.set_debug(debug)
+            return loop.run_until_complete(main)
+        finally:
+            try:
+                _cancel_all_tasks(loop)
+                loop.run_until_complete(loop.shutdown_asyncgens())
+            finally:
+                asyncio.set_event_loop(None)
+                loop.close()
+
+    def _cancel_all_tasks(loop):
+        to_cancel = asyncio.Task.all_tasks(loop)
+        if not to_cancel:
+            return
+        for task in to_cancel:
+            task.cancel()
+        loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True))
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            if task.exception() is not None:
+                loop.call_exception_handler(
+                    {
+                        "message": "unhandled exception during asyncio.run() shutdown",
+                        "exception": task.exception(),
+                        "task": task,
+                    }
+                )
+
+
+if sys.version_info >= (3, 7):
+    asyncio_current_task = asyncio.current_task
+else:
+
+    def asyncio_current_task() -> asyncio.Task:
+        return asyncio.Task.current_task()