From 4191f7d598206ee829376e1a2a4e95ea571de500 Mon Sep 17 00:00:00 2001
From: Eliah Kagan <degeneracypressure@gmail.com>
Date: Thu, 7 Mar 2024 14:29:48 -0500
Subject: [PATCH] Refactor kill_after_timeout logic so mypy can check it

This changes how Git.execute defines the kill_process callback and
how it performs checks, fixing two mypy errors on Windows about how
the signal module doesn't have SIGKILL. In doing so, it also
eliminates the need for the assertion added for safety and clarity
in 2f017ac (#1761), since now kill_process is only defined if it is
to be used (which is also guarded by a platform check, needed by
mypy).

As in dc95a76 before this, part of the change here is to replace
some os.named-based checks with sys.platform-based checks, which is
safe because, when one is specifically checking only for the
distinction between native Windows and all other systems, one can
use either approach. (See dc95a76 for more details on that.)
---
 git/cmd.py | 67 ++++++++++++++++++++++++++++--------------------------
 1 file changed, 35 insertions(+), 32 deletions(-)

diff --git a/git/cmd.py b/git/cmd.py
index 04037c06f..0adc8cf35 100644
--- a/git/cmd.py
+++ b/git/cmd.py
@@ -1134,7 +1134,7 @@ def execute(
         if inline_env is not None:
             env.update(inline_env)
 
-        if os.name == "nt":
+        if sys.platform == "win32":
             cmd_not_found_exception = OSError
             if kill_after_timeout is not None:
                 raise GitCommandError(
@@ -1179,35 +1179,38 @@ def execute(
         if as_process:
             return self.AutoInterrupt(proc, command)
 
-        def kill_process(pid: int) -> None:
-            """Callback to kill a process."""
-            if os.name == "nt":
-                raise AssertionError("Bug: This callback would be ineffective and unsafe on Windows, stopping.")
-            p = Popen(["ps", "--ppid", str(pid)], stdout=PIPE)
-            child_pids = []
-            if p.stdout is not None:
-                for line in p.stdout:
-                    if len(line.split()) > 0:
-                        local_pid = (line.split())[0]
-                        if local_pid.isdigit():
-                            child_pids.append(int(local_pid))
-            try:
-                os.kill(pid, signal.SIGKILL)
-                for child_pid in child_pids:
-                    try:
-                        os.kill(child_pid, signal.SIGKILL)
-                    except OSError:
-                        pass
-                kill_check.set()  # Tell the main routine that the process was killed.
-            except OSError:
-                # It is possible that the process gets completed in the duration after
-                # timeout happens and before we try to kill the process.
-                pass
-            return
-
-        # END kill_process
-
-        if kill_after_timeout is not None:
+        if sys.platform != "win32" and kill_after_timeout is not None:
+
+            def kill_process(pid: int) -> None:
+                """Callback to kill a process.
+
+                This callback implementation would be ineffective and unsafe on Windows.
+                """
+                p = Popen(["ps", "--ppid", str(pid)], stdout=PIPE)
+                child_pids = []
+                if p.stdout is not None:
+                    for line in p.stdout:
+                        if len(line.split()) > 0:
+                            local_pid = (line.split())[0]
+                            if local_pid.isdigit():
+                                child_pids.append(int(local_pid))
+                try:
+                    os.kill(pid, signal.SIGKILL)
+                    for child_pid in child_pids:
+                        try:
+                            os.kill(child_pid, signal.SIGKILL)
+                        except OSError:
+                            pass
+                    # Tell the main routine that the process was killed.
+                    kill_check.set()
+                except OSError:
+                    # It is possible that the process gets completed in the duration
+                    # after timeout happens and before we try to kill the process.
+                    pass
+                return
+
+            # END kill_process
+
             kill_check = threading.Event()
             watchdog = threading.Timer(kill_after_timeout, kill_process, args=(proc.pid,))
 
@@ -1218,10 +1221,10 @@ def kill_process(pid: int) -> None:
         newline = "\n" if universal_newlines else b"\n"
         try:
             if output_stream is None:
-                if kill_after_timeout is not None:
+                if sys.platform != "win32" and kill_after_timeout is not None:
                     watchdog.start()
                 stdout_value, stderr_value = proc.communicate()
-                if kill_after_timeout is not None:
+                if sys.platform != "win32" and kill_after_timeout is not None:
                     watchdog.cancel()
                     if kill_check.is_set():
                         stderr_value = 'Timeout: the command "%s" did not complete in %d ' "secs." % (