diff --git a/.github/workflows/oldest-postgres.yml b/.github/workflows/oldest-postgres.yml
index 0748b175..589899b6 100644
--- a/.github/workflows/oldest-postgres.yml
+++ b/.github/workflows/oldest-postgres.yml
@@ -52,7 +52,7 @@ jobs:
run: |
sudo locale-gen de_DE.UTF-8
- name: install libpq
- if: ${{ contains(inputs.python-versions, 'pypy') }}
+ if: ${{ contains(matrix.python-version, 'pypy') && runner.os == 'Linux' }}
run: sudo apt install libpq5
- name: Install oldest supported versions
uses: fizyk/actions-reuse/.github/actions/pipenv-run@v4.4.4
diff --git a/.github/workflows/single-postgres-windows.yml b/.github/workflows/single-postgres-windows.yml
new file mode 100644
index 00000000..d41d929c
--- /dev/null
+++ b/.github/workflows/single-postgres-windows.yml
@@ -0,0 +1,78 @@
+name: Run pytest tests on Windows
+
+on:
+ workflow_call:
+ inputs:
+ python-versions:
+ description: 'Supported python versions'
+ default: '["3.10", "3.11", "3.12", "3.13", "3.14"]'
+ required: false
+ type: string
+ postgresql:
+ description: 'PostgreSQL version'
+ required: true
+ type: number
+ secrets:
+ codecov_token:
+ description: 'Codecov token'
+ required: false
+
+jobs:
+ postgres:
+ runs-on: windows-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ python-version: ${{ fromJSON(inputs.python-versions) }}
+ env:
+ OS: windows-latest
+ PYTHON: ${{ matrix.python-version }}
+ POSTGRES: ${{ inputs.postgresql }}
+ steps:
+ - uses: actions/checkout@v6
+ - name: Set up Pipenv on python ${{ matrix.python-version }}
+ uses: fizyk/actions-reuse/.github/actions/pipenv-setup@v4.4.4
+ with:
+ python-version: ${{ matrix.python-version }}
+ allow-prereleases: true
+ - uses: ankane/setup-postgres@v1
+ with:
+ postgres-version: ${{ inputs.postgresql }}
+ - name: Detect PostgreSQL path on Windows
+ shell: pwsh
+ run: |
+ $pgPath = "C:\Program Files\PostgreSQL\${{ inputs.postgresql }}\bin\pg_ctl.exe"
+ if (Test-Path $pgPath) {
+ echo "POSTGRESQL_EXEC=$pgPath" >> $env:GITHUB_ENV
+ } else {
+ $pgPath = (Get-Command pg_ctl -ErrorAction SilentlyContinue).Source
+ if ($pgPath) {
+ echo "POSTGRESQL_EXEC=$pgPath" >> $env:GITHUB_ENV
+ }
+ }
+
+ # Verify that PostgreSQL was found
+ if (-not $pgPath) {
+ Write-Error "Error: pg_ctl not found in expected locations. Checked hardcoded path and system PATH."
+ exit 1
+ }
+ - name: Run test
+ uses: fizyk/actions-reuse/.github/actions/pipenv-run@v4.4.4
+ with:
+ command: pytest -svv -p no:xdist --postgresql-exec="${{ env.POSTGRESQL_EXEC }}" -k "not docker" --cov-report=xml --basetemp="${{ runner.temp }}/pytest-basetemp"
+ - name: Run xdist test
+ uses: fizyk/actions-reuse/.github/actions/pipenv-run@v4.4.4
+ with:
+ command: pytest -n auto --dist loadgroup --max-worker-restart 0 --postgresql-exec="${{ env.POSTGRESQL_EXEC }}" -k "not docker" --cov-report=xml:coverage-xdist.xml --basetemp="${{ runner.temp }}/pytest-basetemp"
+ - uses: actions/upload-artifact@v7
+ if: failure()
+ with:
+ name: postgresql-windows-${{ matrix.python-version }}-${{ inputs.postgresql }}
+ path: ${{ runner.temp }}/pytest-basetemp/**
+ - name: Upload coverage to Codecov
+ uses: codecov/codecov-action@v6.0.0
+ with:
+ token: ${{ secrets.codecov_token }}
+ flags: unittests
+ env_vars: OS,PYTHON
+ fail_ci_if_error: false
diff --git a/.github/workflows/single-postgres.yml b/.github/workflows/single-postgres.yml
index a47ea9fc..0c16b5b6 100644
--- a/.github/workflows/single-postgres.yml
+++ b/.github/workflows/single-postgres.yml
@@ -43,6 +43,25 @@ jobs:
- uses: ankane/setup-postgres@v1
with:
postgres-version: ${{ inputs.postgresql }}
+ - name: Set PostgreSQL path
+ run: |
+ # Try to find pg_ctl dynamically for cross-platform compatibility
+ if command -v pg_ctl >/dev/null 2>&1; then
+ PG_CTL_PATH=$(command -v pg_ctl)
+ echo "POSTGRESQL_EXEC=$PG_CTL_PATH" >> $GITHUB_ENV
+ elif [ -f "/opt/homebrew/opt/postgresql@${{ inputs.postgresql }}/bin/pg_ctl" ]; then
+ # macOS Apple Silicon Homebrew path
+ echo "POSTGRESQL_EXEC=/opt/homebrew/opt/postgresql@${{ inputs.postgresql }}/bin/pg_ctl" >> $GITHUB_ENV
+ elif [ -f "/usr/local/opt/postgresql@${{ inputs.postgresql }}/bin/pg_ctl" ]; then
+ # macOS Intel Homebrew path
+ echo "POSTGRESQL_EXEC=/usr/local/opt/postgresql@${{ inputs.postgresql }}/bin/pg_ctl" >> $GITHUB_ENV
+ elif [ -f "/usr/lib/postgresql/${{ inputs.postgresql }}/bin/pg_ctl" ]; then
+ # Debian/Ubuntu path (fallback)
+ echo "POSTGRESQL_EXEC=/usr/lib/postgresql/${{ inputs.postgresql }}/bin/pg_ctl" >> $GITHUB_ENV
+ else
+ echo "Error: pg_ctl not found in expected locations"
+ exit 1
+ fi
- name: Check installed locales
run: |
locale -a
@@ -51,21 +70,21 @@ jobs:
run: |
sudo locale-gen de_DE.UTF-8
- name: install libpq
- if: ${{ contains(inputs.python-versions, 'pypy') }}
+ if: ${{ contains(matrix.python-version, 'pypy') && runner.os == 'Linux' }}
run: sudo apt install libpq5
- name: Run test
uses: fizyk/actions-reuse/.github/actions/pipenv-run@v4.4.4
with:
- command: pytest -svv -p no:xdist --postgresql-exec="/usr/lib/postgresql/${{ inputs.postgresql }}/bin/pg_ctl" -k "not docker" --cov-report=xml
+ command: pytest -svv -p no:xdist --postgresql-exec="${{ env.POSTGRESQL_EXEC }}" -k "not docker" --cov-report=xml --basetemp="${{ runner.temp }}/pytest-basetemp"
- name: Run xdist test
uses: fizyk/actions-reuse/.github/actions/pipenv-run@v4.4.4
with:
- command: pytest -n auto --dist loadgroup --max-worker-restart 0 --postgresql-exec="/usr/lib/postgresql/${{ inputs.postgresql }}/bin/pg_ctl" -k "not docker" --cov-report=xml:coverage-xdist.xml
+ command: pytest -n auto --dist loadgroup --max-worker-restart 0 --postgresql-exec="${{ env.POSTGRESQL_EXEC }}" -k "not docker" --cov-report=xml:coverage-xdist.xml --basetemp="${{ runner.temp }}/pytest-basetemp"
- uses: actions/upload-artifact@v7
if: failure()
with:
name: postgresql-${{ matrix.python-version }}-${{ inputs.postgresql }}
- path: /tmp/pytest-of-runner/**
+ path: ${{ runner.temp }}/pytest-basetemp/**
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v6.0.0
with:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 476d1010..75c3077c 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -77,6 +77,14 @@ jobs:
python-versions: '["3.13", "3.14"]'
secrets:
codecov_token: ${{ secrets.CODECOV_TOKEN }}
+ windows_postgres_17:
+ needs: [postgresql_17]
+ uses: ./.github/workflows/single-postgres-windows.yml
+ with:
+ postgresql: 17
+ python-versions: '["3.12", "3.13", "3.14"]'
+ secrets:
+ codecov_token: ${{ secrets.CODECOV_TOKEN }}
docker_postgresql_18:
needs: [postgresql_18]
uses: ./.github/workflows/dockerised-postgres.yml
diff --git a/newsfragments/1182.feature.rst b/newsfragments/1182.feature.rst
new file mode 100644
index 00000000..29d4f206
--- /dev/null
+++ b/newsfragments/1182.feature.rst
@@ -0,0 +1 @@
+Add Windows support for ``PostgreSQLExecutor``, including platform-specific start/stop handling.
diff --git a/pytest_postgresql/executor.py b/pytest_postgresql/executor.py
index ef027739..186d4b65 100644
--- a/pytest_postgresql/executor.py
+++ b/pytest_postgresql/executor.py
@@ -17,9 +17,12 @@
# along with pytest-postgresql. If not, see .
"""PostgreSQL executor crafter around pg_ctl."""
+import logging
+import os
import os.path
import platform
import re
+import shlex
import shutil
import subprocess
import tempfile
@@ -27,15 +30,21 @@
from typing import Any, Optional, TypeVar
from mirakuru import TCPExecutor
-from mirakuru.exceptions import ProcessFinishedWithError
+from mirakuru.exceptions import ProcessFinishedWithError, TimeoutExpired
from packaging.version import parse
from pytest_postgresql.exceptions import ExecutableMissingException, PostgreSQLUnsupported
+logger = logging.getLogger(__name__)
+
_LOCALE = "C.UTF-8"
if platform.system() == "Darwin":
+ # Darwin does not have C.UTF-8, but en_US.UTF-8 is always available
_LOCALE = "en_US.UTF-8"
+elif platform.system() == "Windows":
+ # Windows doesn't support C.UTF-8 or en_US.UTF-8, use plain "C" locale
+ _LOCALE = "C"
T = TypeVar("T", bound="PostgreSQLExecutor")
@@ -48,17 +57,37 @@ class PostgreSQLExecutor(TCPExecutor):
`_
"""
- BASE_PROC_START_COMMAND = (
- '{executable} start -D "{datadir}" '
+ # Unix command template - uses single quotes for PostgreSQL config value quoting
+ # which protects paths with spaces in unix_socket_directories.
+ # On Unix, mirakuru uses shlex.split() with shell=False, so single quotes
+ # inside double-quoted strings are preserved and passed to PostgreSQL's config parser.
+ UNIX_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}\" "
+ "-c unix_socket_directories='{unixsocketdir}'{postgres_options}\" "
'-l "{logfile}" {startparams}'
)
+ # Windows command template - no single quotes (cmd.exe treats them as literals,
+ # not delimiters) and unix_socket_directories is omitted entirely since PostgreSQL
+ # ignores it on Windows. The -o payload is produced by _windows_pg_options() so
+ # both __init__ and start() always use identical arguments.
+ WINDOWS_PROC_START_COMMAND = '"{executable}" start -D "{datadir}" -o "{pg_options}" -l "{logfile}" {startparams}'
+
VERSION_RE = re.compile(r".* (?P\d+(?:\.\d+)?)")
MIN_SUPPORTED_VERSION = parse("14")
+ @staticmethod
+ def _windows_pg_options(port: int, postgres_options: str) -> str:
+ """Return the -o argument value for the Windows pg_ctl start invocation.
+
+ Centralises the construction so __init__ (command string) and start()
+ (argv list) always produce identical payloads without risk of drift.
+ """
+ extra = f" {postgres_options}" if postgres_options else ""
+ return f"-F -p {port} -c log_destination=stderr -c logging_collector=off{extra}"
+
def __init__(
self,
executable: str,
@@ -108,15 +137,28 @@ def __init__(
self.logfile = logfile
self.startparams = startparams
self.postgres_options = postgres_options
- command = self.BASE_PROC_START_COMMAND.format(
- executable=self.executable,
- datadir=self.datadir,
- port=port,
- unixsocketdir=self.unixsocketdir,
- logfile=self.logfile,
- startparams=self.startparams,
- postgres_options=self.postgres_options,
- )
+ if platform.system() == "Windows":
+ command = self.WINDOWS_PROC_START_COMMAND.format(
+ executable=self.executable,
+ datadir=self.datadir,
+ pg_options=PostgreSQLExecutor._windows_pg_options(port, self.postgres_options),
+ logfile=self.logfile,
+ startparams=self.startparams,
+ )
+ else:
+ # PostgreSQL GUC single-quoted strings double single-quotes to escape them
+ # (e.g. /tmp/o'hare → /tmp/o''hare). Apply this before interpolation so
+ # the generated unix_socket_directories value is always syntactically valid.
+ escaped_unixsocketdir = self.unixsocketdir.replace("'", "''")
+ command = self.UNIX_PROC_START_COMMAND.format(
+ executable=self.executable,
+ datadir=self.datadir,
+ port=port,
+ unixsocketdir=escaped_unixsocketdir,
+ logfile=self.logfile,
+ startparams=self.startparams,
+ postgres_options=f" {self.postgres_options}" if self.postgres_options else "",
+ )
super().__init__(
command,
host,
@@ -145,6 +187,42 @@ def start(self: T) -> T:
f"The currently installed version of PostgreSQL: {self.version}."
)
self.init_directory()
+ if platform.system() == "Windows":
+ # On Windows, pg_ctl start -w exits as soon as the server is
+ # accepting connections. mirakuru's polling loop calls
+ # check_subprocess() (our pg_ctl-status override) repeatedly
+ # while waiting for the launcher subprocess to indicate readiness,
+ # but by the time polling begins the launcher has already exited
+ # so the loop never sees a live process and times out.
+ # Run pg_ctl synchronously as an argv list (no shell) so quoting
+ # is handled safely by subprocess, mirroring the stop() approach.
+ pg_options = self._windows_pg_options(self.port, self.postgres_options)
+ args = [
+ self.executable,
+ "start",
+ "-D",
+ self.datadir,
+ "-o",
+ pg_options,
+ "-l",
+ self.logfile,
+ *shlex.split(self.startparams, posix=False),
+ ]
+ merged_env = os.environ.copy()
+ merged_env.update(self.envvars)
+ try:
+ result = subprocess.run(args, check=False, env=merged_env, timeout=self._timeout)
+ except subprocess.TimeoutExpired as exc:
+ # subprocess.run already killed the stuck pg_ctl process before
+ # re-raising, so no additional cleanup is required here.
+ raise TimeoutExpired(self, self._timeout) from exc
+ if result.returncode != 0:
+ raise ProcessFinishedWithError(self, result.returncode)
+ # pg_ctl start without -w returns before the server accepts connections.
+ # wait_for_postgres() polls self.running() when -w is absent; when -w is
+ # present pg_ctl has already waited so it returns immediately.
+ self.wait_for_postgres()
+ return self
return super().start()
def clean_directory(self) -> None:
@@ -216,20 +294,90 @@ def running(self) -> bool:
"""Check if server is running."""
if not os.path.exists(self.datadir):
return False
- status_code = subprocess.getstatusoutput(f'{self.executable} status -D "{self.datadir}"')[0]
- return status_code == 0
+ result = subprocess.run(
+ [self.executable, "status", "-D", self.datadir],
+ check=False,
+ capture_output=True,
+ )
+ return result.returncode == 0
+
+ def check_subprocess(self) -> bool:
+ """Check whether the PostgreSQL server is ready.
+
+ On Windows the launcher (pg_ctl start -w) is invoked synchronously
+ in start(), so mirakuru's polling loop is never reached and this
+ method is called only from running()-style checks. Returning
+ self.running() (pg_ctl status) is appropriate there.
+
+ On non-Windows, mirakuru's TCPExecutor.check_subprocess() is the
+ correct check: it verifies both that the launcher subprocess is still
+ alive (enabling fast ProcessFinishedWithError on pg_ctl failures) and
+ that the TCP port is accepting connections. Delegating to super()
+ preserves that failure-detection behaviour.
+ """
+ if platform.system() == "Windows":
+ return self.running()
+ return super().check_subprocess()
+
+ def _windows_terminate_process(self, _sig: Optional[int] = None) -> None:
+ """Terminate process on Windows.
+
+ :param _sig: Signal parameter (unused on Windows but included for consistency)
+ """
+ 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()
+ try:
+ self.process.wait(timeout=10)
+ except subprocess.TimeoutExpired:
+ logger.warning(
+ "Process %s could not be cleaned up after kill() and may be a zombie process",
+ self.process.pid if self.process else "unknown",
+ )
+ except (OSError, AttributeError) as e:
+ # Process might already be dead or other issues
+ logger.debug(
+ "Exception during Windows process termination: %s: %s",
+ type(e).__name__,
+ e,
+ )
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)
+ subprocess.check_output(
+ [self.executable, "stop", "-D", self.datadir, "-m", "f"],
+ timeout=self._timeout,
+ )
+ except subprocess.TimeoutExpired as exc:
+ raise TimeoutExpired(self, self._timeout) from exc
+ try:
+ if platform.system() == "Windows":
+ self._windows_terminate_process(sig)
+ # Clean up mirakuru's internal process reference so that
+ # the stopped() context manager can call start() again
+ self._process = None
+ else:
+ super().stop(sig, exp_sig)
except ProcessFinishedWithError:
# Finished, leftovers ought to be cleaned afterwards anyway
pass
+ except AttributeError:
+ # Fallback for edge cases where os.killpg doesn't exist (e.g., Windows)
+ if not hasattr(os, "killpg"):
+ self._windows_terminate_process(sig)
+ self._process = None
+ else:
+ raise
return self
def __del__(self) -> None:
diff --git a/tests/test_executor.py b/tests/test_executor.py
index 25c6aa24..c6944204 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -1,6 +1,8 @@
"""Test various executor behaviours."""
+import platform
from typing import Any
+from unittest.mock import patch
import psycopg
import pytest
@@ -21,13 +23,21 @@ def assert_executor_start_stop(executor: PostgreSQLExecutor) -> None:
"""Check that the executor is working."""
with executor:
assert executor.running()
- psycopg.connect(
- dbname=executor.user,
- user=executor.user,
- password=executor.password,
- host=executor.host,
- port=executor.port,
+ # Retry the connection: under parallel xdist runs the TCP port
+ # becomes accessible before PostgreSQL finishes database recovery,
+ # so an immediate connect may raise OperationalError with
+ # "the database system is starting up".
+ conn = retry(
+ lambda: psycopg.connect(
+ dbname=executor.user,
+ user=executor.user,
+ password=executor.password,
+ host=executor.host,
+ port=executor.port,
+ ),
+ possible_exception=psycopg.OperationalError,
)
+ conn.close()
with pytest.raises(psycopg.OperationalError):
psycopg.connect(
dbname=executor.user,
@@ -106,7 +116,7 @@ def test_executor_init_bad_tmp_path(
pg_exe = process._pg_exe(None, config)
port = process._pg_port(-1, config, [])
tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}") / r"a bad\path/"
- tmpdir.mkdir(exist_ok=True)
+ tmpdir.mkdir(parents=True, exist_ok=True)
datadir, logfile_path = process._prepare_dir(tmpdir, port)
executor = PostgreSQLExecutor(
executable=pg_exe,
@@ -119,6 +129,114 @@ def test_executor_init_bad_tmp_path(
password="some password",
dbname="some database",
)
+
+ # Verify the correct template was selected based on platform
+ current_platform = platform.system()
+ if current_platform == "Windows":
+ # Windows template should not have unix_socket_directories
+ assert "unix_socket_directories" not in executor.command
+ assert "log_destination=stderr" in executor.command
+ else:
+ # Unix/Darwin template should have unix_socket_directories with single quotes
+ assert "unix_socket_directories='" in executor.command
+ assert "log_destination='stderr'" in executor.command
+
+ assert_executor_start_stop(executor)
+
+
+@pytest.mark.parametrize(
+ "platform_name",
+ ["Windows", "Linux", "Darwin", "FreeBSD"],
+)
+def test_executor_platform_template_selection(
+ request: FixtureRequest,
+ tmp_path_factory: pytest.TempPathFactory,
+ platform_name: str,
+) -> None:
+ """Test that correct template is selected for each platform.
+
+ This parametrized test verifies that the executor selects the appropriate
+ command template based on the platform.
+ """
+ config = get_config(request)
+ pg_exe = process._pg_exe(None, config)
+ port = process._pg_port(-1, config, [])
+ tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}")
+ datadir, logfile_path = process._prepare_dir(tmpdir, port)
+
+ with patch("pytest_postgresql.executor.platform.system", return_value=platform_name):
+ executor = PostgreSQLExecutor(
+ executable=pg_exe,
+ host=config.host,
+ port=port,
+ datadir=str(datadir),
+ unixsocketdir=config.unixsocketdir,
+ logfile=str(logfile_path),
+ startparams=config.startparams,
+ dbname="test",
+ )
+
+ # Verify correct template was selected
+ if platform_name == "Windows":
+ # Windows template
+ assert "unix_socket_directories" not in executor.command
+ assert "log_destination=stderr" in executor.command
+ else:
+ # Unix/Darwin template
+ assert "unix_socket_directories='" in executor.command
+ assert "log_destination='stderr'" in executor.command
+
+
+def test_executor_with_special_chars_in_all_paths(
+ request: FixtureRequest,
+ tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+ """Test executor with special characters in multiple paths simultaneously.
+
+ This integration test verifies that the executor can handle special
+ characters (spaces, Unicode) in datadir, logfile, unixsocketdir, and
+ postgres_options all at the same time.
+ """
+ config = get_config(request)
+ pg_exe = process._pg_exe(None, config)
+ port = process._pg_port(-1, config, [])
+ # Create a tmpdir with spaces in the name
+ tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}") / "my test dir"
+ tmpdir.mkdir(exist_ok=True)
+ datadir, logfile_path = process._prepare_dir(tmpdir, port)
+
+ # Create the socket directory for Unix systems.
+ # Use basetemp to keep the path short: Unix domain sockets have a 108-char
+ # OS-level path limit, and the nested test temp path easily exceeds it.
+ socket_dir = tmp_path_factory.getbasetemp() / "sock dir"
+ socket_dir.mkdir(exist_ok=True)
+
+ executor = PostgreSQLExecutor(
+ executable=pg_exe,
+ host=config.host,
+ port=port,
+ datadir=str(datadir),
+ unixsocketdir=str(socket_dir),
+ logfile=str(logfile_path),
+ startparams=config.startparams,
+ password="test pass",
+ dbname="test db",
+ postgres_options="-N 50",
+ )
+
+ # Verify the command contains properly quoted paths
+ command = executor.command
+ assert f'"{datadir}"' in command
+ assert f'"{logfile_path}"' in command
+
+ # Verify correct template was selected based on actual platform
+ current_platform = platform.system()
+ if current_platform == "Windows":
+ assert "unix_socket_directories" not in executor.command
+ else:
+ assert f"unix_socket_directories='{socket_dir}'" in executor.command
+
+ # Start and stop the executor to verify it works
assert_executor_start_stop(executor)
@@ -173,3 +291,119 @@ def test_custom_isolation_level(postgres_isolation_level: Connection) -> None:
cur = postgres_isolation_level.cursor()
cur.execute("SELECT 1")
assert cur.fetchone() == (1,)
+
+
+@pytest.mark.skipif(platform.system() != "Windows", reason="Windows-specific test")
+def test_actual_postgresql_start_windows(
+ request: FixtureRequest,
+ tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+ """Test that PostgreSQL actually starts on Windows with the new template.
+
+ This integration test verifies that the Windows-specific command template
+ correctly starts PostgreSQL on actual Windows systems.
+ """
+ config = get_config(request)
+ pg_exe = process._pg_exe(None, config)
+ port = process._pg_port(-1, config, [])
+ tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}")
+ datadir, logfile_path = process._prepare_dir(tmpdir, port)
+
+ executor = PostgreSQLExecutor(
+ executable=pg_exe,
+ host=config.host,
+ port=port,
+ datadir=str(datadir),
+ unixsocketdir=config.unixsocketdir,
+ logfile=str(logfile_path),
+ startparams=config.startparams,
+ password="testpass",
+ dbname="test",
+ )
+
+ # Verify Windows template is used
+ assert "unix_socket_directories" not in executor.command
+ assert "log_destination=stderr" in executor.command
+
+ # Start and stop PostgreSQL to verify it works
+ assert_executor_start_stop(executor)
+
+
+@pytest.mark.skipif(
+ platform.system() not in ("Linux", "FreeBSD"),
+ reason="Unix/Linux-specific test",
+)
+def test_actual_postgresql_start_unix(
+ request: FixtureRequest,
+ tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+ """Test that PostgreSQL actually starts on Unix/Linux with the new template.
+
+ This integration test verifies that the Unix-specific command template
+ correctly starts PostgreSQL on actual Unix/Linux systems.
+ """
+ config = get_config(request)
+ pg_exe = process._pg_exe(None, config)
+ port = process._pg_port(-1, config, [])
+ tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}")
+ datadir, logfile_path = process._prepare_dir(tmpdir, port)
+
+ executor = PostgreSQLExecutor(
+ executable=pg_exe,
+ host=config.host,
+ port=port,
+ datadir=str(datadir),
+ unixsocketdir=config.unixsocketdir,
+ logfile=str(logfile_path),
+ startparams=config.startparams,
+ password="testpass",
+ dbname="test",
+ )
+
+ # Verify Unix template is used
+ assert "unix_socket_directories='" in executor.command
+ assert "log_destination='stderr'" in executor.command
+
+ # Start and stop PostgreSQL to verify it works
+ assert_executor_start_stop(executor)
+
+
+@pytest.mark.skipif(platform.system() != "Darwin", reason="Darwin/macOS-specific test")
+def test_actual_postgresql_start_darwin(
+ request: FixtureRequest,
+ tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+ """Test that PostgreSQL actually starts on Darwin/macOS with the new template.
+
+ This integration test verifies that the Unix template correctly starts
+ PostgreSQL on actual Darwin/macOS systems and uses the correct locale.
+ """
+ config = get_config(request)
+ pg_exe = process._pg_exe(None, config)
+ port = process._pg_port(-1, config, [])
+ tmpdir = tmp_path_factory.mktemp(f"pytest-postgresql-{request.node.name}")
+ datadir, logfile_path = process._prepare_dir(tmpdir, port)
+
+ executor = PostgreSQLExecutor(
+ executable=pg_exe,
+ host=config.host,
+ port=port,
+ datadir=str(datadir),
+ unixsocketdir=config.unixsocketdir,
+ logfile=str(logfile_path),
+ startparams=config.startparams,
+ password="testpass",
+ dbname="test",
+ )
+
+ # Verify Unix template is used
+ assert "unix_socket_directories='" in executor.command
+ assert "log_destination='stderr'" in executor.command
+
+ # Verify Darwin-specific locale is set
+ assert executor.envvars["LC_ALL"] == "en_US.UTF-8"
+ assert executor.envvars["LC_CTYPE"] == "en_US.UTF-8"
+ assert executor.envvars["LANG"] == "en_US.UTF-8"
+
+ # Start and stop PostgreSQL to verify it works
+ assert_executor_start_stop(executor)
diff --git a/tests/test_postgres_options_plugin.py b/tests/test_postgres_options_plugin.py
index 26e674e0..754dc3b6 100644
--- a/tests/test_postgres_options_plugin.py
+++ b/tests/test_postgres_options_plugin.py
@@ -49,7 +49,10 @@ def test_postgres_loader_in_ini(pointed_pytester: Pytester) -> None:
"""Check that pytest.ini arguments are honored for load."""
pointed_pytester.copy_example("test_load.py")
test_sql_path = pointed_pytester.copy_example("test.sql")
- pointed_pytester.makefile(".ini", pytest=f"[pytest]\npostgresql_load = {test_sql_path}\n")
+ # Use forward slashes so configparser doesn't interpret Windows
+ # backslashes as escape sequences when reading the ini value back
+ sql_path_ini = str(test_sql_path).replace("\\", "/")
+ pointed_pytester.makefile(".ini", pytest=f"[pytest]\npostgresql_load = {sql_path_ini}\n")
ret = pointed_pytester.runpytest("test_load.py")
ret.assert_outcomes(passed=1)
diff --git a/tests/test_windows_compatibility.py b/tests/test_windows_compatibility.py
new file mode 100644
index 00000000..574ea125
--- /dev/null
+++ b/tests/test_windows_compatibility.py
@@ -0,0 +1,894 @@
+"""Test Windows compatibility fixes for pytest-postgresql."""
+
+import os
+import subprocess
+from unittest.mock import MagicMock, patch
+
+from pytest_postgresql.executor import PostgreSQLExecutor
+
+
+class TestCommandTemplates:
+ """Test platform-specific command templates."""
+
+ def test_unix_command_template_has_single_quotes(self) -> None:
+ """Test that Unix template uses single quotes for PostgreSQL config values.
+
+ Single quotes are PostgreSQL config-level quoting that protects paths
+ with spaces in unix_socket_directories. On Unix, mirakuru uses
+ shlex.split() which properly handles single quotes inside double-quoted strings.
+ """
+ template = PostgreSQLExecutor.UNIX_PROC_START_COMMAND
+
+ # Unix template should use single quotes around config values
+ assert "log_destination='stderr'" in template
+ assert "unix_socket_directories='{unixsocketdir}'" in template
+
+ def test_windows_command_template_no_single_quotes(self) -> None:
+ """Test that Windows template has no single quotes.
+
+ Windows cmd.exe treats single quotes as literal characters, not
+ delimiters, which causes errors when passed to pg_ctl.
+ """
+ template = PostgreSQLExecutor.WINDOWS_PROC_START_COMMAND
+
+ # Template itself must contain no single quotes and delegate the -o payload
+ # to the _windows_pg_options helper (tested below).
+ assert "'" not in template
+ assert "log_destination='stderr'" not in template
+
+ # The -o payload produced by the helper must use bare stderr (no single quotes)
+ # because cmd.exe treats single quotes as literal characters.
+ pg_options = PostgreSQLExecutor._windows_pg_options(5432, "")
+ assert "log_destination=stderr" in pg_options
+ assert "'" not in pg_options
+
+ def test_windows_command_template_omits_unix_socket_directories(self) -> None:
+ """Test that Windows template does not include unix_socket_directories.
+
+ PostgreSQL ignores unix_socket_directories on Windows entirely, so
+ including it is unnecessary and avoids any quoting complexity.
+ """
+ template = PostgreSQLExecutor.WINDOWS_PROC_START_COMMAND
+
+ assert "unix_socket_directories" not in template
+ assert "{unixsocketdir}" not in template
+
+ def test_unix_command_template_includes_unix_socket_directories(self) -> None:
+ """Test that Unix template includes unix_socket_directories."""
+ template = PostgreSQLExecutor.UNIX_PROC_START_COMMAND
+
+ assert "unix_socket_directories='{unixsocketdir}'" in template
+
+ def test_unix_template_protects_paths_with_spaces(self) -> None:
+ """Test that Unix template properly quotes paths containing spaces.
+
+ When unixsocketdir contains spaces (e.g., custom temp directories),
+ the single quotes in the Unix template protect the path from being
+ split by PostgreSQL's argument parser.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/my socket dir",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # The path with spaces should be enclosed in single quotes
+ assert "unix_socket_directories='/tmp/my socket dir'" in command
+
+ def test_windows_template_selected_on_windows(self) -> None:
+ """Test that Windows template is selected when platform is Windows."""
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Windows template should not have unix_socket_directories
+ assert "unix_socket_directories" not in command
+ # Windows template should not have single quotes
+ assert "log_destination=stderr" in command
+ assert "log_destination='stderr'" not in command
+
+ def test_unix_template_selected_on_linux(self) -> None:
+ """Test that Unix template is selected when platform is Linux."""
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Unix template should have unix_socket_directories with single quotes
+ assert "unix_socket_directories='/tmp/socket'" in command
+ assert "log_destination='stderr'" in command
+
+ def test_darwin_template_selection(self) -> None:
+ """Test that Darwin/macOS uses Unix template.
+
+ macOS should use the same Unix template as Linux since it's a Unix-like
+ system and supports unix_socket_directories.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Darwin"):
+ executor = PostgreSQLExecutor(
+ executable="/opt/homebrew/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Darwin should use Unix template with unix_socket_directories and single quotes
+ assert "unix_socket_directories='/tmp/socket'" in command
+ assert "log_destination='stderr'" in command
+
+ def test_darwin_locale_setting(self) -> None:
+ """Test that Darwin/macOS sets en_US.UTF-8 locale.
+
+ Darwin requires en_US.UTF-8 instead of C.UTF-8 which is used on Linux.
+ This test verifies the locale environment variables are set correctly.
+ """
+ # Patch the _LOCALE variable to simulate Darwin environment
+ with patch("pytest_postgresql.executor._LOCALE", "en_US.UTF-8"):
+ executor = PostgreSQLExecutor(
+ executable="/opt/homebrew/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Darwin should set en_US.UTF-8 locale
+ assert executor.envvars["LC_ALL"] == "en_US.UTF-8"
+ assert executor.envvars["LC_CTYPE"] == "en_US.UTF-8"
+ assert executor.envvars["LANG"] == "en_US.UTF-8"
+
+ def test_postgres_options_with_single_quotes_unix(self) -> None:
+ """Test postgres_options containing single quotes on Unix.
+
+ Single quotes in postgres_options should be preserved and passed through
+ to PostgreSQL on Unix systems.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ postgres_options="-c shared_buffers='128MB' -c work_mem='64MB'",
+ )
+
+ command = executor.command
+ # postgres_options should be included as-is with single quotes preserved
+ assert "-c shared_buffers='128MB' -c work_mem='64MB'" in command
+
+ def test_postgres_options_with_single_quotes_windows(self) -> None:
+ """Test postgres_options containing single quotes on Windows.
+
+ Single quotes in postgres_options should work on Windows since they're
+ inside the -o parameter's double quotes.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log",
+ startparams="-w",
+ dbname="test",
+ postgres_options="-c shared_buffers='128MB' -c work_mem='64MB'",
+ )
+
+ command = executor.command
+ # postgres_options should be included with single quotes preserved
+ assert "-c shared_buffers='128MB' -c work_mem='64MB'" in command
+
+ def test_postgres_options_with_single_quoted_search_path(self) -> None:
+ """Test postgres_options with a single-quoted search_path value.
+
+ PostgreSQL GUC values are quoted with single quotes in the config syntax.
+ The Unix template wraps the whole -o argument in double quotes, so
+ inner double quotes would prematurely terminate that token and produce an
+ invalid shell command. Single-quoted values are the correct form and must
+ be preserved verbatim inside the -o argument.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ postgres_options="-c search_path='public,other'",
+ )
+
+ command = executor.command
+ # The single-quoted search_path value must appear verbatim inside the -o argument
+ assert "-c search_path='public,other'" in command
+ # Verify it is embedded inside the -o "..." token (no premature " termination)
+ o_start = command.index('-o "')
+ o_end = command.index('"', o_start + 4)
+ o_argument = command[o_start : o_end + 1]
+ assert "-c search_path='public,other'" in o_argument
+
+ def test_postgres_options_with_paths_containing_spaces(self) -> None:
+ """Test postgres_options with file paths containing spaces.
+
+ Config options that reference file paths with spaces should be properly
+ quoted within postgres_options.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ postgres_options="""-c config_file='/etc/postgres/my config.conf'""",
+ )
+
+ command = executor.command
+ # postgres_options with paths containing spaces should be preserved
+ assert """-c config_file='/etc/postgres/my config.conf'""" in command
+
+ def test_empty_postgres_options(self) -> None:
+ """Test command generation with empty postgres_options.
+
+ When postgres_options is empty (default), the command should still be
+ properly formatted without extra spaces or malformed syntax.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ postgres_options="",
+ )
+
+ command = executor.command
+ # Command should still be valid with empty postgres_options
+ assert '"/usr/lib/postgresql/16/bin/pg_ctl" start' in command
+ assert '-D "/tmp/data"' in command
+ assert "unix_socket_directories='/tmp/socket'" in command
+ # Should not have trailing space before closing quote in -o parameter
+ expected_opts = (
+ "-o \"-F -p 5432 -c log_destination='stderr' "
+ "-c logging_collector=off -c unix_socket_directories='/tmp/socket'\""
+ )
+ assert expected_opts in command
+
+ def test_empty_startparams(self) -> None:
+ """Test command generation with empty startparams.
+
+ When startparams is empty (default), the command should still be
+ properly formatted at the end.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Command should be valid with empty startparams
+ assert '"/usr/lib/postgresql/16/bin/pg_ctl" start' in command
+ assert '-l "/tmp/log"' in command
+ # Command should not have trailing spaces at the end
+ assert not command.endswith(" ")
+
+ def test_both_empty_postgres_options_and_startparams(self) -> None:
+ """Test command generation with both postgres_options and startparams empty.
+
+ When both optional parameters are empty, the command should still
+ be properly formatted.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log",
+ startparams="",
+ dbname="test",
+ postgres_options="",
+ )
+
+ command = executor.command
+ # Command should be valid with both empty
+ assert '"C:/Program Files/PostgreSQL/bin/pg_ctl.exe" start' in command
+ assert '-D "C:/temp/data"' in command
+ assert '-l "C:/temp/log"' in command
+ # Windows template should not have unix_socket_directories
+ assert "unix_socket_directories" not in command
+
+ def test_unixsocketdir_ignored_on_windows_in_command(self) -> None:
+ """Test that unixsocketdir value doesn't appear in Windows command.
+
+ Even when unixsocketdir is passed to the executor on Windows, its value
+ should not appear anywhere in the generated command since Windows doesn't
+ use unix_socket_directories.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/this/should/not/appear",
+ logfile="C:/temp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # The unixsocketdir value should NOT appear in the Windows command
+ assert "C:/this/should/not/appear" not in command
+ assert "unix_socket_directories" not in command
+
+ def test_paths_with_multiple_consecutive_spaces(self) -> None:
+ """Test paths with multiple consecutive spaces.
+
+ Paths with multiple spaces should be properly quoted and preserved.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/my socket dir",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Multiple spaces should be preserved
+ assert "unix_socket_directories='/tmp/my socket dir'" in command
+
+ def test_paths_with_special_shell_characters(self) -> None:
+ """Test paths with special shell characters.
+
+ Paths with shell metacharacters should be properly quoted to prevent
+ shell interpretation. Testing with ampersand, semicolon, and pipe.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket&test",
+ logfile="/tmp/log;file",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Special characters should be inside quotes
+ assert "unix_socket_directories='/tmp/socket&test'" in command
+ assert '-l "/tmp/log;file"' in command
+
+ def test_paths_with_unicode_characters(self) -> None:
+ """Test paths with Unicode characters.
+
+ Unicode characters in paths should be properly handled and preserved.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/sóckét_dïr_日本語",
+ logfile="/tmp/lög_文件.log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Unicode characters should be preserved
+ assert "unix_socket_directories='/tmp/sóckét_dïr_日本語'" in command
+ assert '-l "/tmp/lög_文件.log"' in command
+
+ def test_unixsocketdir_with_apostrophe_is_escaped(self) -> None:
+ """Regression test: apostrophes in unixsocketdir are escaped for the GUC parser.
+
+ PostgreSQL single-quoted GUC strings use doubled single-quotes as the
+ escape sequence (SQL-style), so /tmp/o'hare must become
+ unix_socket_directories='/tmp/o''hare'. An unescaped apostrophe would
+ prematurely close the GUC string, making PostgreSQL reject the option or
+ silently misparse it.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/o'hare",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Apostrophe must be doubled so the GUC parser sees a valid string
+ assert "unix_socket_directories='/tmp/o''hare'" in command
+ # Raw un-escaped form must NOT appear
+ assert "unix_socket_directories='/tmp/o'hare'" not in command
+
+ def test_unixsocketdir_with_multiple_apostrophes_are_escaped(self) -> None:
+ """Regression test: multiple apostrophes in unixsocketdir are all escaped."""
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/it's o'hare",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ assert "unix_socket_directories='/tmp/it''s o''hare'" in command
+
+ def test_command_with_all_special_characters_combined(self) -> None:
+ """Test command with multiple types of special characters.
+
+ This comprehensive test combines spaces, quotes, special shell chars,
+ and Unicode to ensure the command handles complex real-world scenarios.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Linux"):
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/16/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/my data & files",
+ unixsocketdir="/tmp/sóckét dir (test)",
+ logfile="/tmp/log file; output.log",
+ startparams="-w -t 30",
+ dbname="test",
+ postgres_options="-c shared_buffers='256MB' -c config_file='/etc/pg/main.conf'",
+ )
+
+ command = executor.command
+ # All special characters should be properly handled
+ assert '-D "/tmp/my data & files"' in command
+ assert "unix_socket_directories='/tmp/sóckét dir (test)'" in command
+ assert '-l "/tmp/log file; output.log"' in command
+ assert "-c shared_buffers='256MB'" in command
+ assert "-c config_file='/etc/pg/main.conf'" in command
+ assert "-w -t 30" in command
+
+
+class TestWindowsCompatibility:
+ """Test Windows-specific process management functionality."""
+
+ def test_windows_terminate_process(self) -> None:
+ """Test Windows process termination."""
+ executor = PostgreSQLExecutor(
+ executable="/path/to/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Mock process
+ mock_process = MagicMock()
+ executor.process = mock_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()
+
+ def test_windows_terminate_process_force_kill(self) -> None:
+ """Test Windows process termination with force kill on timeout."""
+ executor = PostgreSQLExecutor(
+ executable="/path/to/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Mock process that times out
+ mock_process = MagicMock()
+ mock_process.wait.side_effect = [subprocess.TimeoutExpired(cmd="test", timeout=5), None]
+ executor.process = mock_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
+
+ def test_stop_method_windows(self) -> None:
+ """Test stop method on Windows."""
+ executor = PostgreSQLExecutor(
+ executable="/path/to/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Mock subprocess and process
+ with (
+ patch("pytest_postgresql.executor.subprocess.check_output") as mock_subprocess,
+ patch("pytest_postgresql.executor.platform.system", return_value="Windows"),
+ patch.object(executor, "_windows_terminate_process") as mock_terminate,
+ ):
+ result = executor.stop()
+
+ # Should call pg_ctl stop and Windows terminate
+ mock_subprocess.assert_called_once_with(
+ ["/path/to/pg_ctl", "stop", "-D", "/tmp/data", "-m", "f"],
+ )
+ mock_terminate.assert_called_once_with(None)
+ assert result is executor
+
+ def test_stop_method_unix(self) -> None:
+ """Test stop method on Unix systems."""
+ executor = PostgreSQLExecutor(
+ executable="/path/to/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Mock subprocess and super().stop
+ with (
+ patch("pytest_postgresql.executor.subprocess.check_output") as mock_subprocess,
+ patch("pytest_postgresql.executor.platform.system", return_value="Linux"),
+ patch("pytest_postgresql.executor.TCPExecutor.stop") as mock_super_stop,
+ ):
+ mock_super_stop.return_value = executor
+ result = executor.stop()
+
+ # Should call pg_ctl stop and parent class stop
+ mock_subprocess.assert_called_once()
+ mock_super_stop.assert_called_once_with(None, None)
+ assert result is executor
+
+ def test_stop_method_fallback_on_killpg_error(self) -> None:
+ """Test stop method falls back to Windows termination on killpg AttributeError."""
+ executor = PostgreSQLExecutor(
+ executable="/path/to/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ # Mock subprocess and super().stop to raise AttributeError
+ with (
+ patch("pytest_postgresql.executor.subprocess.check_output") as mock_subprocess,
+ patch("pytest_postgresql.executor.platform.system", return_value="Linux"),
+ patch(
+ "pytest_postgresql.executor.TCPExecutor.stop",
+ side_effect=AttributeError("module 'os' has no attribute 'killpg'"),
+ ),
+ patch.object(executor, "_windows_terminate_process") as mock_terminate,
+ ):
+ # Temporarily remove os.killpg so hasattr(os, "killpg") returns False
+ real_killpg = getattr(os, "killpg", None)
+ try:
+ if real_killpg is not None:
+ delattr(os, "killpg")
+ result = executor.stop()
+ finally:
+ if real_killpg is not None:
+ os.killpg = real_killpg
+
+ # Should call pg_ctl stop, fail on super().stop, then use Windows terminate
+ mock_subprocess.assert_called_once()
+ mock_terminate.assert_called_once()
+ assert result is executor
+
+ def test_command_formatting_windows(self) -> None:
+ """Test that command is properly formatted for Windows paths."""
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5555,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log.txt",
+ startparams="-w -s",
+ dbname="testdb",
+ postgres_options="-c shared_preload_libraries=test",
+ )
+
+ # The command should be properly formatted without single quotes
+ # and without unix_socket_directories (irrelevant on Windows)
+ expected_parts = [
+ '"C:/Program Files/PostgreSQL/bin/pg_ctl.exe" start',
+ '-D "C:/temp/data"',
+ '-o "-F -p 5555 -c log_destination=stderr',
+ "-c logging_collector=off",
+ '-c shared_preload_libraries=test"',
+ '-l "C:/temp/log.txt"',
+ "-w -s",
+ ]
+
+ command = executor.command
+ for part in expected_parts:
+ assert part in command, f"Expected '{part}' in command: {command}"
+
+ # Verify unix_socket_directories is NOT in the Windows command
+ assert "unix_socket_directories" not in command, (
+ f"unix_socket_directories should not be in Windows command: {command}"
+ )
+
+ def test_windows_datadir_with_spaces(self) -> None:
+ """Test Windows datadir with spaces in path.
+
+ Windows paths with spaces should be properly quoted with double quotes.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/Program Files/PostgreSQL/my data dir",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # datadir with spaces should be quoted
+ assert '-D "C:/Program Files/PostgreSQL/my data dir"' in command
+
+ def test_windows_logfile_with_spaces(self) -> None:
+ """Test Windows logfile with spaces in path.
+
+ Windows log file paths with spaces should be properly quoted with double quotes.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/Program Files/PostgreSQL/logs/my log file.log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # logfile with spaces should be quoted
+ assert '-l "C:/Program Files/PostgreSQL/logs/my log file.log"' in command
+
+ def test_windows_unc_paths(self) -> None:
+ r"""Test Windows UNC (Universal Naming Convention) paths.
+
+ UNC paths like \\server\share should be properly handled on Windows.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="//server/share/postgres/data",
+ unixsocketdir="//server/share/postgres/socket",
+ logfile="//server/share/postgres/logs/postgresql.log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # UNC paths should be properly quoted (using forward slashes in Python)
+ assert '-D "//server/share/postgres/data"' in command
+ assert '-l "//server/share/postgres/logs/postgresql.log"' in command
+
+ def test_windows_mixed_slashes(self) -> None:
+ """Test Windows paths with mixed forward and backslashes.
+
+ Windows accepts both forward slashes and backslashes, and the command
+ should handle both properly.
+ """
+ with patch("pytest_postgresql.executor.platform.system", return_value="Windows"):
+ executor = PostgreSQLExecutor(
+ executable="C:\\Program Files\\PostgreSQL\\bin\\pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:\\temp\\data",
+ unixsocketdir="C:\\temp\\socket",
+ logfile="C:\\temp\\log.txt",
+ startparams="-w",
+ dbname="test",
+ )
+
+ command = executor.command
+ # Paths with backslashes should be properly quoted
+ assert '"C:\\Program Files\\PostgreSQL\\bin\\pg_ctl.exe" start' in command
+ assert '-D "C:\\temp\\data"' in command
+ assert '-l "C:\\temp\\log.txt"' in command
+
+
+class TestRunningMethod:
+ """Test that running() uses safe list-form subprocess invocation."""
+
+ def test_running_passes_executable_as_argv_element(self) -> None:
+ r"""Test that running() passes the executable as an argv element, not a shell string.
+
+ Using subprocess.run with a list means paths with spaces (e.g.
+ C:\Program Files\...\pg_ctl.exe) are passed as a single token to the
+ OS without any shell parsing, so no quoting is required or desired.
+ """
+ executor = PostgreSQLExecutor(
+ executable="C:/Program Files/PostgreSQL/17/bin/pg_ctl.exe",
+ host="localhost",
+ port=5432,
+ datadir="C:/temp/data",
+ unixsocketdir="C:/temp/socket",
+ logfile="C:/temp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ with (
+ patch("pytest_postgresql.executor.os.path.exists", return_value=True),
+ patch("pytest_postgresql.executor.subprocess.run") as mock_run,
+ ):
+ mock_run.return_value = MagicMock(returncode=0)
+ executor.running()
+
+ args = mock_run.call_args[0][0]
+ assert args[0] == "C:/Program Files/PostgreSQL/17/bin/pg_ctl.exe"
+ assert args[1] == "status"
+ assert args[2] == "-D"
+ assert args[3] == "C:/temp/data"
+
+ def test_running_no_shell_true(self) -> None:
+ """Test that running() does not use shell=True.
+
+ Passing shell=True with a list is both redundant and risky; the list
+ form with shell=False (the default) is the correct approach.
+ """
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/17/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ with (
+ patch("pytest_postgresql.executor.os.path.exists", return_value=True),
+ patch("pytest_postgresql.executor.subprocess.run") as mock_run,
+ ):
+ mock_run.return_value = MagicMock(returncode=0)
+ executor.running()
+
+ kwargs = mock_run.call_args[1]
+ assert kwargs.get("shell", False) is False, "running() must not use shell=True"
+
+ def test_running_returns_true_on_zero_returncode(self) -> None:
+ """Test that running() returns True when pg_ctl status exits 0."""
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/17/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/data",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ with (
+ patch("pytest_postgresql.executor.os.path.exists", return_value=True),
+ patch("pytest_postgresql.executor.subprocess.run") as mock_run,
+ ):
+ mock_run.return_value = MagicMock(returncode=0)
+ assert executor.running() is True
+
+ mock_run.return_value = MagicMock(returncode=3)
+ assert executor.running() is False
+
+ def test_running_datadir_with_spaces_passed_as_argv_element(self) -> None:
+ """Test that a datadir with spaces is passed verbatim as an argv element."""
+ executor = PostgreSQLExecutor(
+ executable="/usr/lib/postgresql/17/bin/pg_ctl",
+ host="localhost",
+ port=5432,
+ datadir="/tmp/my data dir",
+ unixsocketdir="/tmp/socket",
+ logfile="/tmp/log",
+ startparams="-w",
+ dbname="test",
+ )
+
+ with (
+ patch("pytest_postgresql.executor.os.path.exists", return_value=True),
+ patch("pytest_postgresql.executor.subprocess.run") as mock_run,
+ ):
+ mock_run.return_value = MagicMock(returncode=0)
+ executor.running()
+
+ args = mock_run.call_args[0][0]
+ assert args[3] == "/tmp/my data dir", f"Datadir not passed as argv element: {args!r}"