polyfills: os.waitstatus_to_exitcode() for <3.9

src/cockpit/polyfills.py

@@ -4,6 +4,7 @@
 
 
 import contextlib
+import os
 import socket
 
 
@@ -25,6 +26,18 @@ def recv_fds(sock, bufsize, maxfds, flags=0):
 
         socket.recv_fds = recv_fds
 
+    # introduced in 3.9
+    if not hasattr(os, 'waitstatus_to_exitcode'):
+        def waitstatus_to_exitcode(status: int) -> int:
+            if os.WIFEXITED(status):
+                return os.WEXITSTATUS(status)
+            elif os.WIFSIGNALED(status):
+                return -os.WTERMSIG(status)
+            else:
+                raise ValueError(f'Invalid wait status: {status}')
+
+        os.waitstatus_to_exitcode = waitstatus_to_exitcode
+
     # introduced in 3.7
     if not hasattr(contextlib, 'AsyncExitStack'):
         class AsyncExitStack:

src/cockpit/transports.py

@@ -310,15 +310,13 @@ def get_stderr(self, *, reset: bool = False) -> str:
     def watch_exit(self, process: 'subprocess.Popen[bytes]') -> None:
         def child_exited(pid: int, status: int) -> None:
             assert pid == process.pid
-            # os.waitstatus_to_exitcode() is only available since Python 3.9
-            if os.WIFEXITED(status):
-                self._returncode = os.WEXITSTATUS(status)
-            elif os.WIFSIGNALED(status):
-                self._returncode = -os.WTERMSIG(status)
-            else:
+            assert isinstance(self._protocol, SubprocessProtocol)
+
+            try:
+                self._returncode = os.waitstatus_to_exitcode(status)
+            except ValueError:
                 self._returncode = status
 
-            assert isinstance(self._protocol, SubprocessProtocol)
             logger.debug('Process exited with status %d', self._returncode)
             if not self._closing:
                 self._protocol.process_exited()