subt.launch: add generic process launcher

subt/launch.py

@@ -0,0 +1,51 @@
+import logging
+import platform
+import signal
+import subprocess
+
+g_logger = logging.getLogger(__name__)
+
+class Launch:
+    def __init__(self, config, bus):
+        self.bus = bus
+        self.command = config['command']
+        self.shell = config.get('shell', False)
+
+    def start(self):
+        self.running = subprocess.Popen(self.command, shell=self.shell)
+
+    def request_stop(self):
+        self.bus.shutdown()
+        signum = signal.CTRL_C_EVENT if platform.system() == "Windows" else signal.SIGINT
+        self.running.send_signal(signum)
+
+    def join(self, timeout=None):
+        try:
+            self.running.wait(timeout)
+        except subprocess.TimeoutExpired:
+            command = self.command if isinstance(self.command, str) else " ".join(self.command)
+            g_logger.warning(f"'{command}' still running, terminating")
+            self.running.terminate()
+            try:
+                self.running.wait(1)
+            except subprocess.TimeoutExpired:
+                g_logger.warning(f"'{command}' ignoring SIGTERM, killing")
+                self.running.kill()
+                self.running.wait()
+        assert self.running.poll() is not None
+
+
+if __name__ == "__main__":
+    from unittest.mock import MagicMock
+    test_launch = Launch(config={'command': ['echo', 'hello world!']}, bus=MagicMock())
+    test_launch.start()
+    test_launch.join()
+
+    test_request_stop = Launch(config={'command': ['sleep', '10']}, bus=MagicMock())
+    test_request_stop.start()
+    test_request_stop.request_stop()
+    test_request_stop.join(0.1)
+
+    test_join = Launch(config={'command': ['sleep', '10']}, bus=MagicMock())
+    test_join.start()
+    test_join.join(0.1)

subt/test_launch.py

@@ -0,0 +1,32 @@
+import logging
+import unittest
+import unittest.mock
+
+from . import launch
+
+
+class Test(unittest.TestCase):
+    def test_launch(self):
+        p = launch.Launch(config={'command': ['true']}, bus=unittest.mock.MagicMock())
+        p.start()
+        p.join()
+
+    def test_request_stop(self):
+        p = launch.Launch(config={'command': ['sleep', '10']}, bus=unittest.mock.MagicMock())
+        p.start()
+        p.request_stop()
+        p.join(0.1)
+
+    def test_join(self):
+        with self.assertLogs(level=logging.WARNING) as log:
+            p = launch.Launch(config={'command': ['sleep', '10']}, bus=unittest.mock.MagicMock())
+            p.start()
+            p.join(0.1)
+        self.assertEqual(len(log.records), 1)
+
+    def test_shell(self):
+        with self.assertLogs(level=logging.WARNING) as log:
+            p = launch.Launch(config={'command': 'sleep 10', 'shell': True}, bus=unittest.mock.MagicMock())
+            p.start()
+            p.join(0.1)
+        self.assertEqual(len(log.records), 1)