Refactor PostgreSQLExecutor for cross-platform command compatibility

- Updated _get_base_command method to provide a unified command format for both Windows and Unix systems, eliminating the use of single quotes around configuration values.
- Simplified process termination logic by removing platform checks from the _windows_terminate_process method, ensuring consistent behavior across operating systems.
- Enhanced tests to verify that the base command format is consistent across platforms.

Author: tboy1337

pytest_postgresql/executor.py

@@ -51,25 +51,17 @@ class PostgreSQLExecutor(TCPExecutor):
     """
 
     def _get_base_command(self) -> str:
-        """Get the base PostgreSQL command, Windows-compatible."""
-        if platform.system() == "Windows":
-            # Windows doesn't handle single quotes well in subprocess calls
-            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}'
-            )
-        else:
-            # Unix/Linux systems work with single 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}'
-            )
+        """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
 
@@ -242,20 +234,28 @@ def _windows_terminate_process(self, sig: Optional[int] = None) -> None:
             return
 
         try:
-            if platform.system() == "Windows":
-                # 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()
-            else:
-                # On Unix systems, use the signal
-                actual_sig = sig or signal.SIGTERM
-                os.killpg(self.process.pid, actual_sig)
+            # 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

tests/test_windows_compatibility.py

@@ -9,8 +9,8 @@
 class TestWindowsCompatibility:
     """Test Windows-specific functionality."""
 
-    def test_get_base_command_windows(self):
-        """Test that Windows base command doesn't use single quotes."""
+    def test_get_base_command_unified(self):
+        """Test that base command is unified across platforms."""
         executor = PostgreSQLExecutor(
             executable="/path/to/pg_ctl",
             host="localhost",
@@ -22,32 +22,21 @@ def test_get_base_command_windows(self):
             dbname="test",
         )
 
+        # Test that command format is consistent across platforms
         with patch("platform.system", return_value="Windows"):
-            command = executor._get_base_command()
-            # Windows command should not have single quotes around stderr
-            assert "log_destination=stderr" in command
-            assert "log_destination='stderr'" not in command
-            assert "unix_socket_directories={unixsocketdir}" in command
-            assert "unix_socket_directories='{unixsocketdir}'" not in command
-
-    def test_get_base_command_unix(self):
-        """Test that Unix base command uses single quotes."""
-        executor = PostgreSQLExecutor(
-            executable="/path/to/pg_ctl",
-            host="localhost",
-            port=5432,
-            datadir="/tmp/data",
-            unixsocketdir="/tmp/socket",
-            logfile="/tmp/log",
-            startparams="-w",
-            dbname="test",
-        )
+            windows_command = executor._get_base_command()
 
         with patch("platform.system", return_value="Linux"):
-            command = executor._get_base_command()
-            # Unix command should have single quotes around stderr
-            assert "log_destination='stderr'" in command
-            assert "unix_socket_directories='{unixsocketdir}'" in command
+            unix_command = executor._get_base_command()
+
+        # Both should be the same now
+        assert windows_command == unix_command
+        
+        # Both should use the simplified format without single quotes
+        assert "log_destination=stderr" in windows_command
+        assert "log_destination='stderr'" not in windows_command
+        assert "unix_socket_directories={unixsocketdir}" in windows_command
+        assert "unix_socket_directories='{unixsocketdir}'" not in windows_command
 
     def test_windows_terminate_process(self):
         """Test Windows process termination."""
@@ -66,12 +55,12 @@ def test_windows_terminate_process(self):
         mock_process = MagicMock()
         executor.process = mock_process
 
-        with patch("platform.system", return_value="Windows"):
-            executor._windows_terminate_process()
+        # No need to mock platform.system() since the method doesn't check it anymore
+        executor._windows_terminate_process()
 
-            # Should call terminate first
-            mock_process.terminate.assert_called_once()
-            mock_process.wait.assert_called()
+        # Should call terminate first
+        mock_process.terminate.assert_called_once()
+        mock_process.wait.assert_called()
 
     def test_windows_terminate_process_force_kill(self):
         """Test Windows process termination with force kill on timeout."""
@@ -91,13 +80,13 @@ def test_windows_terminate_process_force_kill(self):
         mock_process.wait.side_effect = [subprocess.TimeoutExpired(cmd="test", timeout=5), None]
         executor.process = mock_process
 
-        with patch("platform.system", return_value="Windows"):
-            executor._windows_terminate_process()
+        # No need to mock platform.system() since the method doesn't check it anymore
+        executor._windows_terminate_process()
 
-            # Should call terminate, wait (timeout), then kill, then wait again
-            mock_process.terminate.assert_called_once()
-            mock_process.kill.assert_called_once()
-            assert mock_process.wait.call_count == 2
+        # Should call terminate, wait (timeout), then kill, then wait again
+        mock_process.terminate.assert_called_once()
+        mock_process.kill.assert_called_once()
+        assert mock_process.wait.call_count == 2
 
     def test_stop_method_windows(self):
         """Test stop method on Windows."""