feat: add Windows compatibility fixes for PostgreSQL executor

Author: tboy1337

pytest_postgresql/executor.py

@@ -17,10 +17,12 @@
 # along with pytest-postgresql.  If not, see <http://www.gnu.org/licenses/>.
 """PostgreSQL executor crafter around pg_ctl."""
 
+import os
 import os.path
 import platform
 import re
 import shutil
+import signal
 import subprocess
 import tempfile
 import time
@@ -48,13 +50,20 @@ class PostgreSQLExecutor(TCPExecutor):
     <http://www.postgresql.org/docs/current/static/app-pg-ctl.html>`_
     """
 
-    BASE_PROC_START_COMMAND = (
-        '{executable} start -D "{datadir}" '
-        "-o \"-F -p {port} -c log_destination='stderr' "
-        "-c logging_collector=off "
-        "-c unix_socket_directories='{unixsocketdir}' {postgres_options}\" "
-        '-l "{logfile}" {startparams}'
-    )
+    def _get_base_command(self) -> str:
+        """Get the base PostgreSQL command, cross-platform compatible."""
+        # Use unified format without single quotes around values
+        # This format works on both Windows and Unix systems since PostgreSQL
+        # configuration values without spaces don't require quotes
+        return (
+            '{executable} start -D "{datadir}" '
+            '-o "-F -p {port} -c log_destination=stderr '
+            "-c logging_collector=off "
+            '-c unix_socket_directories={unixsocketdir} {postgres_options}" '
+            '-l "{logfile}" {startparams}'
+        )
+
+    BASE_PROC_START_COMMAND = ""  # Will be set dynamically
 
     VERSION_RE = re.compile(r".* (?P<version>\d+(?:\.\d+)?)")
     MIN_SUPPORTED_VERSION = parse("14")
@@ -108,7 +117,7 @@ def __init__(
         self.logfile = logfile
         self.startparams = startparams
         self.postgres_options = postgres_options
-        command = self.BASE_PROC_START_COMMAND.format(
+        command = self._get_base_command().format(
             executable=self.executable,
             datadir=self.datadir,
             port=port,
@@ -219,17 +228,58 @@ def running(self) -> bool:
         status_code = subprocess.getstatusoutput(f'{self.executable} status -D "{self.datadir}"')[0]
         return status_code == 0
 
+    def _windows_terminate_process(self, sig: Optional[int] = None) -> None:
+        """Terminate process on Windows."""
+        if self.process is None:
+            return
+
+        try:
+            # On Windows, try to terminate gracefully first
+            self.process.terminate()
+            # Give it a chance to terminate gracefully
+            try:
+                self.process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                # If it doesn't terminate gracefully, force kill
+                self.process.kill()
+                self.process.wait()
+        except (OSError, AttributeError):
+            # Process might already be dead or other issues
+            pass
+
+    def _unix_terminate_process(self, sig: Optional[int] = None) -> None:
+        """Terminate process on Unix systems."""
+        if self.process is None:
+            return
+
+        try:
+            # On Unix systems, use the signal
+            actual_sig = sig or signal.SIGTERM
+            os.killpg(self.process.pid, actual_sig)
+        except (OSError, AttributeError):
+            # Process might already be dead or other issues
+            pass
+
     def stop(self: T, sig: Optional[int] = None, exp_sig: Optional[int] = None) -> T:
         """Issue a stop request to executable."""
         subprocess.check_output(
             f'{self.executable} stop -D "{self.datadir}" -m f',
             shell=True,
         )
         try:
-            super().stop(sig, exp_sig)
+            if platform.system() == "Windows":
+                self._windows_terminate_process(sig)
+            else:
+                super().stop(sig, exp_sig)
         except ProcessFinishedWithError:
             # Finished, leftovers ought to be cleaned afterwards anyway
             pass
+        except AttributeError as e:
+            # Handle case where os.killpg doesn't exist (shouldn't happen now)
+            if "killpg" in str(e):
+                self._windows_terminate_process(sig)
+            else:
+                raise
         return self
 
     def __del__(self) -> None: