test_pypeline_daemon: fix flaky SIGSEGV

Avoid triggering a flaky SIGSEGV caused by running the asyncio event
loop in a thread. Switch to using `Process` which avoids the issue
altogether.

tests/pypeline/python/daemon/base.py

@@ -29,6 +29,9 @@ class TestServer(ABC):
         self.pipebox_path = str(self.tmp_dir_path / "pipebox.py")
         self.pipeline_path = str(self.tmp_dir_path / "pipeline.yml")
 
+    def stop(self):
+        pass
+
     @abstractmethod
     def get_epoch(self) -> Response: ...
 

tests/pypeline/python/daemon/starlette_daemon.py

@@ -4,8 +4,9 @@
 
 import logging
 import socket
-import threading
 import time
+from multiprocessing import Process
+from signal import SIGTERM
 from tempfile import TemporaryDirectory
 
 import requests
@@ -34,7 +35,6 @@ class StarletteTestServer(TestServer):
     def __init__(self, port=None, storage_provider_url: str = "memory://"):
         super().__init__()
         self.port = port or find_free_port()
-        self.server_thread = None
         self.base_url = f"http://127.0.0.1:{self.port}"
         self.project_id = "test_project_id"
         self.storage_provider_url = storage_provider_url
@@ -46,16 +46,26 @@ class StarletteTestServer(TestServer):
         self.session = requests.Session()
         self.session.headers["X-Projectid"] = self.project_id
 
-        # The server thread has to be a daemon so when the tests finish it will
+        # The server daemon has to be a daemon so when the tests finish it will
         # be killed. But this silences the exceptions, so if it fails, you need
         # to run it manually to fix them.
-        self.server_thread = threading.Thread(target=self._run_server, daemon=True)
-        self.server_thread.start()
+        self.server_process: Process | None = Process(target=self._run_server, daemon=True)
+        self.server_process.start()
         self._wait_for_server()
 
+    def __del__(self):
+        self.stop()
+
+    def stop(self):
+        if self.server_process is not None:
+            self.server_process.terminate()
+            self.server_process.join()
+            assert self.server_process.exitcode in (0, -SIGTERM)
+            self.server_process = None
+
     def _run_server(self):
         """
-        This is the thread that calls the cli to spawn the daemon.
+        This runs in a separate process that calls the cli to spawn the daemon.
         Beware that if it raises an exception it will not be print on stdout or
         stderr, so if the daemon doesn't start, try to run it manually.
         """

tests/pypeline/python/test_daemon.py

@@ -38,14 +38,14 @@ def storage_provider_url(request):
 def daemon_server(request, storage_provider_url):
     """Fixture that provides a running daemon server for testing."""
     if request.param == "json":
-        return JsonTestServer(
-            storage_provider_url=storage_provider_url,
-        )
+        instance = JsonTestServer(storage_provider_url=storage_provider_url)
     elif request.param == "starlette":
-        return StarletteTestServer(
-            storage_provider_url=storage_provider_url,
-        )
-    raise ValueError(f"Unknown daemon type {request.param}")
+        instance = StarletteTestServer(storage_provider_url=storage_provider_url)
+    else:
+        raise ValueError(f"Unknown daemon type {request.param}")
+
+    yield instance
+    instance.stop()
 
 
 def test_daemon(daemon_server: TestServer):