public inbox for [email protected]
help / color / mirror / Atom feedFrom: Jelte Fennema-Nio <[email protected]>
To: Aleksander Alekseev <[email protected]>
To: PostgreSQL Hackers <[email protected]>
Cc: Andres Freund <[email protected]>
Cc: Jacob Champion <[email protected]>
Cc: Robert Haas <[email protected]>
Cc: Daniel Gustafsson <[email protected]>
Cc: Tom Lane <[email protected]>
Cc: Peter Eisentraut <[email protected]>
Cc: Nazir Bilal Yavuz <[email protected]>
Subject: Re: RFC: adding pytest as a supported test framework
Date: Sun, 01 Mar 2026 23:16:38 +0100
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <CAOYmi+nEqA2LmetcJKUDmctypPLLumkVwj3vQ3idYd8yAGza5Q@mail.gmail.com>
<CAOYmi+kebAt6wSX7ee0c0kMzV7r0hp93bAt10V5a88yHHUKwog@mail.gmail.com>
<CAOYmi+=CrBFBRX-foRRES2tx2wXBJJhbsJGjgFbWvPcmBJuK-Q@mail.gmail.com>
<[email protected]>
<[email protected]>
<g4gdtfedwwdgu5sbcopjt3djtqk6p2q7n5nymp7ppapfwukoyd@ev2v4jtsbure>
<CAOYmi+mV97+FC=ME0EZj+3LBFHZ3g_xGmTGA_3u+TwJ_pvpFWw@mail.gmail.com>
<[email protected]>
<baqtdyuunu42yu7ler4bflifksznt7u7tywj4atdtcxwxhxinj@76taubczsku2>
<CAGECzQTzG0otdj7Z6--FOtbLLiceJ+eWn3dqRYr6-toG+RWcEg@mail.gmail.com>
<froy7mvxq76o2asuyzajyvaqelzuv372xsylfvhlhkvis7ojr4@23lg5adaxvdw>
<[email protected]>
<[email protected]>
<CAJ7c6TMt-aXXmtO9JgXEsh4UvWrPQbW4_CnRxftAacW0QM9a5g@mail.gmail.com>
<[email protected]>
On Sun Jan 25, 2026 at 1:54 PM CET, Jelte Fennema-Nio wrote:
>> I haven't looked at 0004 and 0005.
>
> Makes sense. I mostly kept those patches as part of the patchset to make
> sure I did not break Jacob's usecase with all my changes to the test
> suite. They were more meant as: "I did not touch these patches, nor are
> they ready to merge, but it's possible to do stuff like this too".
These patches caused CI to fail after the addition of the protocol
grease (4966bd3ed95). I fixed the tests (by forcing usage of 3.2), but I
attached them as nocfbot now, so that they can't be the reason that CI
fails again.
Attachments:
[text/x-patch] nocfbot-v12-0004-WIP-pytest-Add-some-SSL-client-tests.patch (22.6K, 2-nocfbot-v12-0004-WIP-pytest-Add-some-SSL-client-tests.patch)
download | inline diff:
From 953512a17aede111f2c9a77dc92fd1dbc3365145 Mon Sep 17 00:00:00 2001
From: Jacob Champion <[email protected]>
Date: Tue, 16 Dec 2025 09:30:55 +0100
Subject: [PATCH v12 4/5] WIP: pytest: Add some SSL client tests
This is a sample client-only test suite. It tests some handshake
failures against a mock server, as well as a full SSL handshake + empty
query + response.
pyca/cryptography is added as a new package dependency. Certificates for
testing are generated on the fly.
The mock design is threaded: the server socket is listening on a
background thread, and the test provides the server logic via a
callback. There is some additional work still needed to make this
production-ready; see the notes for _TCPServer.background(). (Currently,
an exception in the wrong place could result in a hang-until-timeout
rather than an immediate failure.)
TODOs:
- local_server and tcp_server_class are nearly identical and should
share code.
- fix exception-related timeouts for .background()
- figure out the proper use of "session" vs "module" scope
- ensure that pq.libpq unwinds (to close connections) before tcp_server;
see comment in test_server_with_ssl_disabled()
---
.cirrus.tasks.yml | 2 +
pyproject.toml | 8 +
src/test/pytest/pyt/test_libpq.py | 141 +++++++++++++++
src/test/ssl/Makefile | 2 +
src/test/ssl/meson.build | 6 +
src/test/ssl/pyt/conftest.py | 128 ++++++++++++++
src/test/ssl/pyt/test_client.py | 280 ++++++++++++++++++++++++++++++
7 files changed, 567 insertions(+)
create mode 100644 src/test/ssl/pyt/conftest.py
create mode 100644 src/test/ssl/pyt/test_client.py
diff --git a/.cirrus.tasks.yml b/.cirrus.tasks.yml
index 47ec9def20e..e3459c2e08b 100644
--- a/.cirrus.tasks.yml
+++ b/.cirrus.tasks.yml
@@ -647,6 +647,7 @@ task:
CIRRUS_WORKING_DIR: ${HOME}/pgsql/
CCACHE_DIR: ${HOME}/ccache
MACPORTS_CACHE: ${HOME}/macports-cache
+ PYTEST_DEBUG_TEMPROOT: /tmp # default is too long for UNIX sockets on Mac
MESON_FEATURES: >-
-Dbonjour=enabled
@@ -667,6 +668,7 @@ task:
p5.34-io-tty
p5.34-ipc-run
python312
+ py312-cryptography
py312-packaging
py312-pytest
tcl
diff --git a/pyproject.toml b/pyproject.toml
index 4628d2274e0..00c8ae88583 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,6 +12,14 @@ dependencies = [
# Any other dependencies are effectively optional (added below). We import
# these libraries using pytest.importorskip(). So tests will be skipped if
# they are not available.
+
+ # Notes on the cryptography package:
+ # - 3.3.2 is shipped on Debian bullseye.
+ # - 3.4.x drops support for Python 2, making it a version of note for older LTS
+ # distros.
+ # - 35.x switched versioning schemes and moved to Rust parsing.
+ # - 40.x is the last version supporting Python 3.6.
+ "cryptography >= 3.3.2",
]
[tool.pytest.ini_options]
diff --git a/src/test/pytest/pyt/test_libpq.py b/src/test/pytest/pyt/test_libpq.py
index 1d0d9bc3b94..01b2761a76f 100644
--- a/src/test/pytest/pyt/test_libpq.py
+++ b/src/test/pytest/pyt/test_libpq.py
@@ -33,3 +33,144 @@ def test_must_connect_errors(connect):
"""Tests that connect() raises LibpqError."""
with pytest.raises(LibpqError, match="invalid connection option"):
connect(some_unknown_keyword="whatever")
+
+
[email protected]
+def local_server(tmp_path, remaining_timeout):
+ """
+ Opens up a local UNIX socket for mocking a Postgres server on a background
+ thread. See the _Server API for usage.
+
+ This fixture requires AF_UNIX support; dependent tests will be skipped on
+ platforms that don't provide it.
+ """
+
+ try:
+ from socket import AF_UNIX
+ except ImportError:
+ pytest.skip("AF_UNIX not supported on this platform")
+
+ class _Server(contextlib.ExitStack):
+ """
+ Implementation class for local_server. See .background() for the primary
+ entry point for tests. Postgres clients may connect to this server via
+ local_server.host/local_server.port.
+
+ _Server derives from contextlib.ExitStack to provide easy cleanup of
+ associated resources; see the documentation for that class for a full
+ explanation.
+ """
+
+ def __init__(self):
+ super().__init__()
+
+ self.host = tmp_path
+ self.port = 5432
+
+ self._thread = None
+ self._thread_exc = None
+ self._listener = self.enter_context(
+ socket.socket(AF_UNIX, socket.SOCK_STREAM),
+ )
+
+ def bind_and_listen(self):
+ """
+ Does the actual work of binding the UNIX socket using the Postgres
+ server conventions and listening for connections.
+
+ The listen backlog is currently hardcoded to one.
+ """
+ sockfile = self.host / ".s.PGSQL.{}".format(self.port)
+
+ # Lock down the permissions on the new socket.
+ prev_mask = os.umask(0o077)
+
+ # Bind (creating the socket file), and immediately register it for
+ # deletion from disk when the stack is cleaned up.
+ self._listener.bind(bytes(sockfile))
+ self.callback(os.unlink, sockfile)
+
+ os.umask(prev_mask)
+
+ self._listener.listen(1)
+
+ def background(self, fn: Callable[[socket.socket], None]) -> None:
+ """
+ Accepts a client connection on a background thread and passes it to
+ the provided callback. Any exceptions raised from the callback will
+ be re-raised on the main thread during fixture teardown.
+
+ Blocking operations on the connected socket default to using the
+ remaining_timeout(), though this can be changed by the test via the
+ socket's .settimeout().
+ """
+
+ def _bg():
+ try:
+ self._listener.settimeout(remaining_timeout())
+ sock, _ = self._listener.accept()
+
+ with sock:
+ sock.settimeout(remaining_timeout())
+ fn(sock)
+
+ except Exception as e:
+ # Save the exception for re-raising on the main thread.
+ self._thread_exc = e
+
+ # TODO: rather than using callback(), consider explicitly signaling
+ # the fn() implementation to stop early if we get an exception.
+ # Otherwise we'll hang until the end of the timeout.
+ self._thread = threading.Thread(target=_bg)
+ self.callback(self._join)
+
+ self._thread.start()
+
+ def _join(self):
+ """
+ Waits for the background thread to finish and raises any thrown
+ exception. This is called during fixture teardown.
+ """
+ # Give a little bit of wiggle room on the join timeout, since we're
+ # racing against the test's own use of remaining_timeout(). (It's
+ # preferable to let tests report timeouts; the stack traces will
+ # help with debugging.)
+ self._thread.join(remaining_timeout() + 1)
+ if self._thread.is_alive():
+ raise TimeoutError("background thread is still running after timeout")
+
+ if self._thread_exc is not None:
+ raise self._thread_exc
+
+ with _Server() as s:
+ s.bind_and_listen()
+ yield s
+
+
+def test_connection_is_finished_on_error(connect, local_server):
+ """Tests that PQfinish() gets called at the end of testing."""
+ expected_error = "something is wrong"
+
+ def serve_error(s: socket.socket) -> None:
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+
+ # Quick check for the startup packet version.
+ version = struct.unpack("!HH", s.recv(4))
+ assert version == (3, 2)
+
+ # Discard the remainder of the startup packet and send a v2 error.
+ s.recv(pktlen - 8)
+ s.send(b"E" + expected_error.encode() + b"\0")
+
+ # And now the socket should be closed.
+ assert not s.recv(1), "client sent unexpected data"
+
+ local_server.background(serve_error)
+
+ with pytest.raises(LibpqError, match=expected_error):
+ # Exiting this context should result in PQfinish().
+ connect(
+ host=local_server.host,
+ port=local_server.port,
+ max_protocol_version="3.2", # Don't use grease
+ )
diff --git a/src/test/ssl/Makefile b/src/test/ssl/Makefile
index aa062945fb9..287729ad9fb 100644
--- a/src/test/ssl/Makefile
+++ b/src/test/ssl/Makefile
@@ -30,6 +30,8 @@ clean distclean:
# Doesn't depend on sslfiles because we don't rebuild them by default
check:
$(prove_check)
+ # XXX these suites should run independently, not serially
+ $(pytest_check)
installcheck:
$(prove_installcheck)
diff --git a/src/test/ssl/meson.build b/src/test/ssl/meson.build
index 9e5bdbb6136..6ec274d8165 100644
--- a/src/test/ssl/meson.build
+++ b/src/test/ssl/meson.build
@@ -15,4 +15,10 @@ tests += {
't/003_sslinfo.pl',
],
},
+ 'pytest': {
+ 'tests': [
+ 'pyt/test_client.py',
+ 'pyt/test_server.py',
+ ],
+ },
}
diff --git a/src/test/ssl/pyt/conftest.py b/src/test/ssl/pyt/conftest.py
new file mode 100644
index 00000000000..870f738ac44
--- /dev/null
+++ b/src/test/ssl/pyt/conftest.py
@@ -0,0 +1,128 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import datetime
+import re
+import subprocess
+import tempfile
+from collections import namedtuple
+
+import pytest
+
+
[email protected](scope="session")
+def cryptography():
+ return pytest.importorskip("cryptography", "3.3.2")
+
+
+Cert = namedtuple("Cert", "cert, certpath, key, keypath")
+
+
[email protected](scope="session")
+def certs(cryptography, tmp_path_factory):
+ """
+ Caches commonly used certificates at the session level, and provides a way
+ to create new ones.
+
+ - certs.ca: the root CA certificate
+
+ - certs.server: the "standard" server certficate, signed by certs.ca
+
+ - certs.server_host: the hostname of the certs.server certificate
+
+ - certs.new(): creates a custom certificate, signed by certs.ca
+ """
+
+ from cryptography import x509
+ from cryptography.hazmat.primitives import hashes, serialization
+ from cryptography.hazmat.primitives.asymmetric import rsa
+ from cryptography.x509.oid import NameOID
+
+ tmpdir = tmp_path_factory.mktemp("test-certs")
+
+ class _Certs:
+ def __init__(self):
+ self.ca = self.new(
+ x509.Name(
+ [x509.NameAttribute(NameOID.COMMON_NAME, "PG pytest CA")],
+ ),
+ ca=True,
+ )
+
+ self.server_host = "example.org"
+ self.server = self.new(
+ x509.Name(
+ [x509.NameAttribute(NameOID.COMMON_NAME, self.server_host)],
+ )
+ )
+
+ def new(self, subject: x509.Name, *, ca=False) -> Cert:
+ """
+ Creates and signs a new Cert with the given subject name. If ca is
+ True, the certificate will be self-signed; otherwise the certificate
+ is signed by self.ca.
+ """
+ key = rsa.generate_private_key(
+ public_exponent=65537,
+ key_size=2048,
+ )
+
+ builder = x509.CertificateBuilder()
+ now = datetime.datetime.now(datetime.timezone.utc)
+
+ builder = (
+ builder.subject_name(subject)
+ .public_key(key.public_key())
+ .serial_number(x509.random_serial_number())
+ .not_valid_before(now)
+ .not_valid_after(now + datetime.timedelta(hours=1))
+ )
+
+ if ca:
+ builder = builder.issuer_name(subject)
+ else:
+ builder = builder.issuer_name(self.ca.cert.subject)
+
+ builder = builder.add_extension(
+ x509.BasicConstraints(ca=ca, path_length=None),
+ critical=True,
+ )
+
+ cert = builder.sign(
+ private_key=key if ca else self.ca.key,
+ algorithm=hashes.SHA256(),
+ )
+
+ # Dump the certificate and key to file.
+ keypath = self._tofile(
+ key.private_bytes(
+ serialization.Encoding.PEM,
+ serialization.PrivateFormat.PKCS8,
+ serialization.NoEncryption(),
+ ),
+ suffix=".key",
+ )
+ certpath = self._tofile(
+ cert.public_bytes(serialization.Encoding.PEM),
+ suffix="-ca.crt" if ca else ".crt",
+ )
+
+ return Cert(
+ cert=cert,
+ certpath=certpath,
+ key=key,
+ keypath=keypath,
+ )
+
+ def _tofile(self, data: bytes, *, suffix) -> str:
+ """
+ Dumps data to a file on disk with the requested suffix and returns
+ the path. The file is located somewhere in pytest's temporary
+ directory root.
+ """
+ f = tempfile.NamedTemporaryFile(suffix=suffix, dir=tmpdir, delete=False)
+ with f:
+ f.write(data)
+
+ return f.name
+
+ return _Certs()
diff --git a/src/test/ssl/pyt/test_client.py b/src/test/ssl/pyt/test_client.py
new file mode 100644
index 00000000000..4113dd21752
--- /dev/null
+++ b/src/test/ssl/pyt/test_client.py
@@ -0,0 +1,280 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import contextlib
+import ctypes
+import socket
+import ssl
+import struct
+import threading
+from typing import Callable
+
+import pytest
+
+import pypg
+from libpq import LibpqError, ExecStatus
+
+# This suite opens up local TCP ports and is hidden behind PG_TEST_EXTRA=ssl.
+pytestmark = pypg.require_test_extras("ssl")
+
+
[email protected](scope="session", autouse=True)
+def skip_if_no_ssl_support(libpq_handle):
+ """Skips tests if SSL support is not configured."""
+
+ # Declare PQsslAttribute().
+ PQsslAttribute = libpq_handle.PQsslAttribute
+ PQsslAttribute.restype = ctypes.c_char_p
+ PQsslAttribute.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
+
+ if not PQsslAttribute(None, b"library"):
+ pytest.skip("requires SSL support to be configured")
+
+
+#
+# Test Fixtures
+#
+
+
[email protected]
+def tcp_server_class(remaining_timeout):
+ """
+ Metafixture to combine related logic for tcp_server and ssl_server.
+
+ TODO: combine with test_libpq.local_server
+ """
+
+ class _TCPServer(contextlib.ExitStack):
+ """
+ Implementation class for tcp_server. See .background() for the primary
+ entry point for tests. Postgres clients may connect to this server via
+ **tcp_server.conninfo.
+
+ _TCPServer derives from contextlib.ExitStack to provide easy cleanup of
+ associated resources; see the documentation for that class for a full
+ explanation.
+ """
+
+ def __init__(self):
+ super().__init__()
+
+ self._thread = None
+ self._thread_exc = None
+ self._listener = self.enter_context(
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
+ )
+
+ self._bind_and_listen()
+ sockname = self._listener.getsockname()
+ self.conninfo = dict(
+ hostaddr=sockname[0],
+ port=sockname[1],
+ )
+
+ def _bind_and_listen(self):
+ """
+ Does the actual work of binding the socket and listening for
+ connections.
+
+ The listen backlog is currently hardcoded to one.
+ """
+ self._listener.bind(("127.0.0.1", 0))
+ self._listener.listen(1)
+
+ def background(self, fn: Callable[[socket.socket], None]) -> None:
+ """
+ Accepts a client connection on a background thread and passes it to
+ the provided callback. Any exceptions raised from the callback will
+ be re-raised on the main thread during fixture teardown.
+
+ Blocking operations on the connected socket default to using the
+ remaining_timeout(), though this can be changed by the test via the
+ socket's .settimeout().
+ """
+
+ def _bg():
+ try:
+ self._listener.settimeout(remaining_timeout())
+ sock, _ = self._listener.accept()
+
+ with sock:
+ sock.settimeout(remaining_timeout())
+ fn(sock)
+
+ except Exception as e:
+ # Save the exception for re-raising on the main thread.
+ self._thread_exc = e
+
+ # TODO: rather than using callback(), consider explicitly signaling
+ # the fn() implementation to stop early if we get an exception.
+ # Otherwise we'll hang until the end of the timeout.
+ self._thread = threading.Thread(target=_bg)
+ self.callback(self._join)
+
+ self._thread.start()
+
+ def _join(self):
+ """
+ Waits for the background thread to finish and raises any thrown
+ exception. This is called during fixture teardown.
+ """
+ # Give a little bit of wiggle room on the join timeout, since we're
+ # racing against the test's own use of remaining_timeout(). (It's
+ # preferable to let tests report timeouts; the stack traces will
+ # help with debugging.)
+ self._thread.join(remaining_timeout() + 1)
+ if self._thread.is_alive():
+ raise TimeoutError("background thread is still running after timeout")
+
+ if self._thread_exc is not None:
+ raise self._thread_exc
+
+ return _TCPServer
+
+
[email protected]
+def tcp_server(tcp_server_class):
+ """
+ Opens up a local TCP socket for mocking a Postgres server on a background
+ thread. See the _TCPServer API for usage.
+ """
+ with tcp_server_class() as s:
+ yield s
+
+
[email protected]
+def ssl_server(tcp_server_class, certs):
+ """
+ Like tcp_server, but with an additional .background_ssl() method which will
+ perform a SSLRequest handshake on the socket before handing the connection
+ to the test callback.
+
+ This server uses certs.server as its identity.
+ """
+
+ class _SSLServer(tcp_server_class):
+ def __init__(self):
+ super().__init__()
+
+ self.conninfo["host"] = certs.server_host
+
+ self._ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ self._ctx.load_cert_chain(certs.server.certpath, certs.server.keypath)
+
+ def background_ssl(self, fn: Callable[[ssl.SSLSocket], None]) -> None:
+ """
+ Invokes a server callback as with .background(), but an SSLRequest
+ handshake is performed first, and the socket provided to the
+ callback has been wrapped in an OpenSSL layer.
+ """
+
+ def handshake(s: socket.socket):
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+
+ # Make sure we get an SSLRequest.
+ version = struct.unpack("!HH", s.recv(4))
+ assert version == (1234, 5679)
+ assert pktlen == 8
+
+ # Accept the SSLRequest.
+ s.send(b"S")
+
+ with self._ctx.wrap_socket(s, server_side=True) as wrapped:
+ fn(wrapped)
+
+ self.background(handshake)
+
+ with _SSLServer() as s:
+ yield s
+
+
+#
+# Tests
+#
+
+
[email protected]("sslmode", ("require", "verify-ca", "verify-full"))
+def test_server_with_ssl_disabled(connect, tcp_server, certs, sslmode):
+ """
+ Make sure client refuses to talk to non-SSL servers with stricter
+ sslmodes.
+ """
+
+ def refuse_ssl(s: socket.socket):
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+
+ # Make sure we get an SSLRequest.
+ version = struct.unpack("!HH", s.recv(4))
+ assert version == (1234, 5679)
+ assert pktlen == 8
+
+ # Refuse the SSLRequest.
+ s.send(b"N")
+
+ # Wait for the client to close the connection.
+ assert not s.recv(1), "client sent unexpected data"
+
+ tcp_server.background(refuse_ssl)
+
+ with pytest.raises(LibpqError, match="server does not support SSL"):
+ connect(
+ **tcp_server.conninfo,
+ sslrootcert=certs.ca.certpath,
+ sslmode=sslmode,
+ max_protocol_version="3.2", # Don't use grease
+ )
+
+
+def test_verify_full_connection(connect, ssl_server, certs):
+ """Completes a verify-full connection and empty query."""
+
+ def handle_empty_query(s: ssl.SSLSocket):
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+
+ # Check the startup packet version, then discard the remainder.
+ version = struct.unpack("!HH", s.recv(4))
+ assert version == (3, 2)
+ s.recv(pktlen - 8)
+
+ # Send the required litany of server messages.
+ s.send(struct.pack("!cII", b"R", 8, 0)) # AuthenticationOK
+
+ # ParameterStatus: client_encoding
+ key = b"client_encoding\0"
+ val = b"UTF-8\0"
+ s.send(struct.pack("!cI", b"S", 4 + len(key) + len(val)) + key + val)
+
+ # ParameterStatus: DateStyle
+ key = b"DateStyle\0"
+ val = b"ISO, MDY\0"
+ s.send(struct.pack("!cI", b"S", 4 + len(key) + len(val)) + key + val)
+
+ s.send(struct.pack("!cIII", b"K", 12, 1234, 1234)) # BackendKeyData
+ s.send(struct.pack("!cIc", b"Z", 5, b"I")) # ReadyForQuery
+
+ # Expect an empty query.
+ pkttype = s.recv(1)
+ assert pkttype == b"Q"
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+ assert s.recv(pktlen - 4) == b"\0"
+
+ # Send an EmptyQueryResponse+ReadyForQuery.
+ s.send(struct.pack("!cI", b"I", 4))
+ s.send(struct.pack("!cIc", b"Z", 5, b"I"))
+
+ # libpq should terminate and close the connection.
+ assert s.recv(1) == b"X"
+ pktlen = struct.unpack("!I", s.recv(4))[0]
+ assert pktlen == 4
+
+ assert not s.recv(1), "client sent unexpected data"
+
+ ssl_server.background_ssl(handle_empty_query)
+
+ conn = connect(
+ **ssl_server.conninfo,
+ sslrootcert=certs.ca.certpath,
+ sslmode="verify-full",
+ max_protocol_version="3.2", # Don't use grease
+ )
+ with conn:
+ assert conn.exec("").status() == ExecStatus.PGRES_EMPTY_QUERY
--
2.53.0
[text/x-patch] nocfbot-v12-0005-WIP-pytest-Add-some-server-side-SSL-tests.patch (8.5K, 3-nocfbot-v12-0005-WIP-pytest-Add-some-server-side-SSL-tests.patch)
download | inline diff:
From 8874627bee0590e4a7de96ba37bc8f6e66fb6c39 Mon Sep 17 00:00:00 2001
From: Jacob Champion <[email protected]>
Date: Tue, 16 Dec 2025 09:31:46 +0100
Subject: [PATCH v12 5/5] WIP: pytest: Add some server-side SSL tests
In the same vein as the previous commit, this is a server-only test
suite operating against a mock client. The test itself is a heavily
parameterized check for direct-SSL handshake behavior, using a
combination of "standard" and "custom" certificates via the certs
fixture.
installcheck is currently unsupported, but the architecture has some
extension points that should make it possible later. For now, a new
server is always started for the test session.
TODOs:
- improve remaining_timeout() integration with socket operations; at the
moment, the timeout resets on every call rather than decrementing
---
src/test/ssl/pyt/conftest.py | 50 ++++++++++
src/test/ssl/pyt/test_server.py | 161 ++++++++++++++++++++++++++++++++
2 files changed, 211 insertions(+)
create mode 100644 src/test/ssl/pyt/test_server.py
diff --git a/src/test/ssl/pyt/conftest.py b/src/test/ssl/pyt/conftest.py
index 870f738ac44..d121724800b 100644
--- a/src/test/ssl/pyt/conftest.py
+++ b/src/test/ssl/pyt/conftest.py
@@ -126,3 +126,53 @@ def certs(cryptography, tmp_path_factory):
return f.name
return _Certs()
+
+
[email protected](scope="module", autouse=True)
+def ssl_setup(pg_server_module, certs, datadir):
+ """
+ Sets up required server settings for all tests in this module.
+ """
+ try:
+ with pg_server_module.restarting() as s:
+ s.conf.set(
+ ssl="on",
+ ssl_ca_file=certs.ca.certpath,
+ ssl_cert_file=certs.server.certpath,
+ ssl_key_file=certs.server.keypath,
+ )
+
+ # Reject by default.
+ s.hba.prepend("hostssl all all all reject")
+
+ except subprocess.CalledProcessError:
+ # This is a decent place to skip if the server isn't set up for SSL.
+ logpath = datadir / "postgresql.log"
+ unsupported = re.compile("SSL is not supported")
+
+ with open(logpath, "r") as log:
+ for line in log:
+ if unsupported.search(line):
+ pytest.skip("the server does not support SSL")
+
+ # Some other error happened.
+ raise
+
+ users = pg_server_module.create_users("ssl")
+ dbs = pg_server_module.create_dbs("ssl")
+
+ return (users, dbs)
+
+
[email protected](scope="module")
+def client_cert(ssl_setup, certs):
+ """
+ Creates a Cert for the "ssl" user.
+ """
+ from cryptography import x509
+ from cryptography.x509.oid import NameOID
+
+ users, _ = ssl_setup
+ user = users["ssl"]
+
+ return certs.new(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, user)]))
diff --git a/src/test/ssl/pyt/test_server.py b/src/test/ssl/pyt/test_server.py
new file mode 100644
index 00000000000..d5cb14b6c9a
--- /dev/null
+++ b/src/test/ssl/pyt/test_server.py
@@ -0,0 +1,161 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import re
+import socket
+import ssl
+import struct
+
+import pytest
+
+import pypg
+
+# This suite opens up local TCP ports and is hidden behind PG_TEST_EXTRA=ssl.
+pytestmark = pypg.require_test_extras("ssl")
+
+# For use with the `creds` parameter below.
+CLIENT = "client"
+SERVER = "server"
+
+
+# fmt: off
[email protected](
+ "auth_method, creds, expected_error",
+[
+ # Trust allows anything.
+ ("trust", None, None),
+ ("trust", CLIENT, None),
+ ("trust", SERVER, None),
+
+ # verify-ca allows any CA-signed certificate.
+ ("trust clientcert=verify-ca", None, "requires a valid client certificate"),
+ ("trust clientcert=verify-ca", CLIENT, None),
+ ("trust clientcert=verify-ca", SERVER, None),
+
+ # cert and verify-full allow only the correct certificate.
+ ("trust clientcert=verify-full", None, "requires a valid client certificate"),
+ ("trust clientcert=verify-full", CLIENT, None),
+ ("trust clientcert=verify-full", SERVER, "authentication failed for user"),
+ ("cert", None, "requires a valid client certificate"),
+ ("cert", CLIENT, None),
+ ("cert", SERVER, "authentication failed for user"),
+],
+)
+# fmt: on
+def test_direct_ssl_certificate_authentication(
+ pg,
+ ssl_setup,
+ certs,
+ client_cert,
+ remaining_timeout,
+ # test parameters
+ auth_method,
+ creds,
+ expected_error,
+):
+ """
+ Tests direct SSL connections with various client-certificate/HBA
+ combinations.
+ """
+
+ # Set up the HBA as desired by the test.
+ users, dbs = ssl_setup
+
+ user = users["ssl"]
+ db = dbs["ssl"]
+
+ with pg.reloading() as s:
+ s.hba.prepend(
+ ["hostssl", db, user, "127.0.0.1/32", auth_method],
+ ["hostssl", db, user, "::1/128", auth_method],
+ )
+
+ # Configure the SSL settings for the client.
+ ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+ ctx.load_verify_locations(cafile=certs.ca.certpath)
+ ctx.set_alpn_protocols(["postgresql"]) # for direct SSL
+
+ # Load up a client certificate if required by the test.
+ if creds == CLIENT:
+ ctx.load_cert_chain(client_cert.certpath, client_cert.keypath)
+ elif creds == SERVER:
+ # Using a server certificate as the client credential is expected to
+ # work only for clientcert=verify-ca (and `trust`, naturally).
+ ctx.load_cert_chain(certs.server.certpath, certs.server.keypath)
+
+ # Make a direct SSL connection. There's no SSLRequest in the handshake; we
+ # simply wrap a TCP connection with OpenSSL.
+ addr = (pg.hostaddr, pg.port)
+ with socket.create_connection(addr) as s:
+ s.settimeout(remaining_timeout()) # XXX this resets every operation
+
+ with ctx.wrap_socket(s, server_hostname=certs.server_host) as conn:
+ # Build and send the startup packet.
+ startup_options = dict(
+ user=user,
+ database=db,
+ application_name="pytest",
+ )
+
+ payload = b""
+ for k, v in startup_options.items():
+ payload += k.encode() + b"\0"
+ payload += str(v).encode() + b"\0"
+ payload += b"\0" # null terminator
+
+ pktlen = 4 + 4 + len(payload)
+ conn.send(struct.pack("!IHH", pktlen, 3, 0) + payload)
+
+ if not expected_error:
+ # Expect an AuthenticationOK to come back.
+ pkttype, pktlen = struct.unpack("!cI", conn.recv(5))
+ assert pkttype == b"R"
+ assert pktlen == 8
+
+ authn_result = struct.unpack("!I", conn.recv(4))[0]
+ assert authn_result == 0
+
+ # Read and discard to ReadyForQuery.
+ while True:
+ pkttype, pktlen = struct.unpack("!cI", conn.recv(5))
+ payload = conn.recv(pktlen - 4)
+
+ if pkttype == b"Z":
+ assert payload == b"I"
+ break
+
+ # Send an empty query.
+ conn.send(struct.pack("!cI", b"Q", 5) + b"\0")
+
+ # Expect EmptyQueryResponse+ReadyForQuery.
+ pkttype, pktlen = struct.unpack("!cI", conn.recv(5))
+ assert pkttype == b"I"
+ assert pktlen == 4
+
+ pkttype, pktlen = struct.unpack("!cI", conn.recv(5))
+ assert pkttype == b"Z"
+
+ payload = conn.recv(pktlen - 4)
+ assert payload == b"I"
+
+ else:
+ # Match the expected authentication error.
+ pkttype, pktlen = struct.unpack("!cI", conn.recv(5))
+ assert pkttype == b"E"
+
+ payload = conn.recv(pktlen - 4)
+ msg = None
+
+ for component in payload.split(b"\0"):
+ if not component:
+ break # end of message
+
+ key, val = component[:1], component[1:]
+ if key == b"S":
+ assert val == b"FATAL"
+ elif key == b"M":
+ msg = val.decode()
+
+ assert re.search(expected_error, msg), "server error did not match"
+
+ # Terminate.
+ conn.send(struct.pack("!cI", b"X", 4))
--
2.53.0
[text/x-patch] v12-0001-Add-support-for-pytest-test-suites.patch (27.8K, 4-v12-0001-Add-support-for-pytest-test-suites.patch)
download | inline diff:
From a0e1c24a222191bd382e517b8c5d4b4f81664085 Mon Sep 17 00:00:00 2001
From: Jacob Champion <[email protected]>
Date: Wed, 13 Aug 2025 10:58:56 -0700
Subject: [PATCH v12 1/5] Add support for pytest test suites
Specify --enable-pytest/-Dpytest=enabled at configure time. This
contains no Postgres test logic -- it is just a "vanilla" pytest
skeleton.
This contains a custom pytest plugin to generate TAP output. This plugin
is used by the Meson mtest runner, to show relevant information for
failed tests. The pytest-tap plugin would have been preferable, but it's
now in maintenance mode, and it has problems with accidentally
suppressing important collection failures.
Co-authored-by: Jelte Fennema-Nio <[email protected]>
---
.cirrus.tasks.yml | 11 +-
.gitignore | 3 +
configure | 166 +++++++++++++++++++++++++++++-
configure.ac | 24 ++++-
meson.build | 100 ++++++++++++++++++
meson_options.txt | 8 +-
pyproject.toml | 21 ++++
src/Makefile.global.in | 29 ++++++
src/makefiles/meson.build | 2 +
src/test/Makefile | 1 +
src/test/meson.build | 1 +
src/test/pytest/Makefile | 20 ++++
src/test/pytest/README | 1 +
src/test/pytest/meson.build | 15 +++
src/test/pytest/pgtap.py | 197 ++++++++++++++++++++++++++++++++++++
src/tools/testwrap | 6 +-
16 files changed, 597 insertions(+), 8 deletions(-)
create mode 100644 pyproject.toml
create mode 100644 src/test/pytest/Makefile
create mode 100644 src/test/pytest/README
create mode 100644 src/test/pytest/meson.build
create mode 100644 src/test/pytest/pgtap.py
diff --git a/.cirrus.tasks.yml b/.cirrus.tasks.yml
index 4841a204248..47ec9def20e 100644
--- a/.cirrus.tasks.yml
+++ b/.cirrus.tasks.yml
@@ -44,6 +44,7 @@ env:
-Dldap=enabled
-Dssl=openssl
-Dtap_tests=enabled
+ -Dpytest=enabled
-Dplperl=enabled
-Dplpython=enabled
-Ddocs=enabled
@@ -316,6 +317,7 @@ task:
-Dlibcurl=enabled
-Dnls=enabled
-Dpam=enabled
+ -DPYTEST=pytest-3.12
setup_additional_packages_script: |
#pkgin -y install ...
@@ -519,14 +521,15 @@ task:
set -e
./configure \
--enable-cassert --enable-injection-points --enable-debug \
- --enable-tap-tests --enable-nls \
+ --enable-tap-tests --enable-pytest --enable-nls \
--with-segsize-blocks=6 \
--with-libnuma \
--with-liburing \
\
${LINUX_CONFIGURE_FEATURES} \
\
- CLANG="ccache clang"
+ CLANG="ccache clang" \
+ PYTEST="env LD_PRELOAD=/lib/x86_64-linux-gnu/libasan.so.8 pytest"
EOF
build_script: su postgres -c "make -s -j${BUILD_JOBS} world-bin"
upload_caches: ccache
@@ -664,6 +667,8 @@ task:
p5.34-io-tty
p5.34-ipc-run
python312
+ py312-packaging
+ py312-pytest
tcl
zstd
@@ -713,6 +718,7 @@ task:
sh src/tools/ci/ci_macports_packages.sh $MACOS_PACKAGE_LIST
# system python doesn't provide headers
sudo /opt/local/bin/port select python3 python312
+ sudo /opt/local/bin/port select pytest pytest312
# Make macports install visible for subsequent steps
echo PATH=/opt/local/sbin/:/opt/local/bin/:$PATH >> $CIRRUS_ENV
upload_caches: macports
@@ -786,6 +792,7 @@ task:
-Dldap=enabled
-Dssl=openssl
-Dtap_tests=enabled
+ -Dpytest=enabled
-Dplperl=enabled
-Dplpython=enabled
diff --git a/.gitignore b/.gitignore
index 4e911395fe3..a550ce6194b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -31,6 +31,7 @@ win32ver.rc
*.exe
lib*dll.def
lib*.pc
+__pycache__/
# Local excludes in root directory
/GNUmakefile
@@ -43,3 +44,5 @@ lib*.pc
/Release/
/tmp_install/
/portlock/
+/.venv/
+/uv.lock
diff --git a/configure b/configure
index a285a6ec3d7..b6e32cf4549 100755
--- a/configure
+++ b/configure
@@ -630,6 +630,8 @@ vpath_build
PG_SYSROOT
PG_VERSION_NUM
LDFLAGS_EX_BE
+UV
+PYTEST
PROVE
DBTOEPUB
FOP
@@ -773,6 +775,7 @@ CFLAGS
CC
enable_injection_points
PG_TEST_EXTRA
+enable_pytest
enable_tap_tests
enable_dtrace
DTRACEFLAGS
@@ -851,6 +854,7 @@ enable_profiling
enable_coverage
enable_dtrace
enable_tap_tests
+enable_pytest
enable_injection_points
with_blocksize
with_segsize
@@ -1551,7 +1555,10 @@ Optional Features:
--enable-profiling build with profiling enabled
--enable-coverage build with coverage testing instrumentation
--enable-dtrace build with DTrace support
- --enable-tap-tests enable TAP tests (requires Perl and IPC::Run)
+ --enable-tap-tests enable (Perl-based) TAP tests (requires Perl and
+ IPC::Run)
+ --enable-pytest enable (Python-based) pytest suites (requires
+ Python)
--enable-injection-points
enable injection points (for testing)
--enable-depend turn on automatic dependency tracking
@@ -3634,7 +3641,7 @@ fi
#
-# TAP tests
+# Test frameworks
#
@@ -3662,6 +3669,32 @@ fi
+
+# Check whether --enable-pytest was given.
+if test "${enable_pytest+set}" = set; then :
+ enableval=$enable_pytest;
+ case $enableval in
+ yes)
+ :
+ ;;
+ no)
+ :
+ ;;
+ *)
+ as_fn_error $? "no argument expected for --enable-pytest option" "$LINENO" 5
+ ;;
+ esac
+
+else
+ enable_pytest=no
+
+fi
+
+
+
+
+
+
#
# Injection points
#
@@ -19261,6 +19294,135 @@ $as_echo "$modulestderr" >&6; }
fi
fi
+if test "$enable_pytest" = yes; then
+ if test -z "$PYTEST"; then
+ for ac_prog in pytest py.test
+do
+ # Extract the first word of "$ac_prog", so it can be a program name with args.
+set dummy $ac_prog; ac_word=$2
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking for $ac_word" >&5
+$as_echo_n "checking for $ac_word... " >&6; }
+if ${ac_cv_path_PYTEST+:} false; then :
+ $as_echo_n "(cached) " >&6
+else
+ case $PYTEST in
+ [\\/]* | ?:[\\/]*)
+ ac_cv_path_PYTEST="$PYTEST" # Let the user override the test with a path.
+ ;;
+ *)
+ as_save_IFS=$IFS; IFS=$PATH_SEPARATOR
+for as_dir in $PATH
+do
+ IFS=$as_save_IFS
+ test -z "$as_dir" && as_dir=.
+ for ac_exec_ext in '' $ac_executable_extensions; do
+ if as_fn_executable_p "$as_dir/$ac_word$ac_exec_ext"; then
+ ac_cv_path_PYTEST="$as_dir/$ac_word$ac_exec_ext"
+ $as_echo "$as_me:${as_lineno-$LINENO}: found $as_dir/$ac_word$ac_exec_ext" >&5
+ break 2
+ fi
+done
+ done
+IFS=$as_save_IFS
+
+ ;;
+esac
+fi
+PYTEST=$ac_cv_path_PYTEST
+if test -n "$PYTEST"; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: $PYTEST" >&5
+$as_echo "$PYTEST" >&6; }
+else
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+fi
+
+
+ test -n "$PYTEST" && break
+done
+
+else
+ # Report the value of PYTEST in configure's output in all cases.
+ { $as_echo "$as_me:${as_lineno-$LINENO}: checking for PYTEST" >&5
+$as_echo_n "checking for PYTEST... " >&6; }
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: $PYTEST" >&5
+$as_echo "$PYTEST" >&6; }
+fi
+
+ if test -z "$PYTEST"; then
+ # If pytest not found, try installing with uv
+ if test -z "$UV"; then
+ for ac_prog in uv
+do
+ # Extract the first word of "$ac_prog", so it can be a program name with args.
+set dummy $ac_prog; ac_word=$2
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking for $ac_word" >&5
+$as_echo_n "checking for $ac_word... " >&6; }
+if ${ac_cv_path_UV+:} false; then :
+ $as_echo_n "(cached) " >&6
+else
+ case $UV in
+ [\\/]* | ?:[\\/]*)
+ ac_cv_path_UV="$UV" # Let the user override the test with a path.
+ ;;
+ *)
+ as_save_IFS=$IFS; IFS=$PATH_SEPARATOR
+for as_dir in $PATH
+do
+ IFS=$as_save_IFS
+ test -z "$as_dir" && as_dir=.
+ for ac_exec_ext in '' $ac_executable_extensions; do
+ if as_fn_executable_p "$as_dir/$ac_word$ac_exec_ext"; then
+ ac_cv_path_UV="$as_dir/$ac_word$ac_exec_ext"
+ $as_echo "$as_me:${as_lineno-$LINENO}: found $as_dir/$ac_word$ac_exec_ext" >&5
+ break 2
+ fi
+done
+ done
+IFS=$as_save_IFS
+
+ ;;
+esac
+fi
+UV=$ac_cv_path_UV
+if test -n "$UV"; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: $UV" >&5
+$as_echo "$UV" >&6; }
+else
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+fi
+
+
+ test -n "$UV" && break
+done
+
+else
+ # Report the value of UV in configure's output in all cases.
+ { $as_echo "$as_me:${as_lineno-$LINENO}: checking for UV" >&5
+$as_echo_n "checking for UV... " >&6; }
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: $UV" >&5
+$as_echo "$UV" >&6; }
+fi
+
+ if test -n "$UV"; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: checking whether uv can install pytest dependencies" >&5
+$as_echo_n "checking whether uv can install pytest dependencies... " >&6; }
+ if "$UV" pip install "$srcdir" >&5 2>&1; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: yes" >&5
+$as_echo "yes" >&6; }
+ PYTEST="$UV run pytest"
+ else
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+ as_fn_error $? "pytest not found and uv failed to install dependencies" "$LINENO" 5
+ fi
+ else
+ as_fn_error $? "pytest not found" "$LINENO" 5
+ fi
+ fi
+fi
+
# If compiler will take -Wl,--as-needed (or various platform-specific
# spellings thereof) then add that to LDFLAGS. This is much easier than
# trying to filter LIBS to the minimum for each executable.
diff --git a/configure.ac b/configure.ac
index 476a76c7991..dbf9bd98f95 100644
--- a/configure.ac
+++ b/configure.ac
@@ -226,11 +226,16 @@ AC_SUBST(DTRACEFLAGS)])
AC_SUBST(enable_dtrace)
#
-# TAP tests
+# Test frameworks
#
PGAC_ARG_BOOL(enable, tap-tests, no,
- [enable TAP tests (requires Perl and IPC::Run)])
+ [enable (Perl-based) TAP tests (requires Perl and IPC::Run)])
AC_SUBST(enable_tap_tests)
+
+PGAC_ARG_BOOL(enable, pytest, no,
+ [enable (Python-based) pytest suites (requires Python)])
+AC_SUBST(enable_pytest)
+
AC_ARG_VAR(PG_TEST_EXTRA,
[enable selected extra tests (overridden at runtime by PG_TEST_EXTRA environment variable)])
@@ -2457,6 +2462,21 @@ if test "$enable_tap_tests" = yes; then
fi
fi
+if test "$enable_pytest" = yes; then
+ PGAC_PATH_PROGS(PYTEST, [pytest py.test])
+ if test -z "$PYTEST"; then
+ # Try python -m pytest as a fallback
+ AC_MSG_CHECKING([whether python -m pytest works])
+ if "$PYTHON" -m pytest --version >&AS_MESSAGE_LOG_FD 2>&1; then
+ AC_MSG_RESULT([yes])
+ PYTEST="$PYTHON -m pytest"
+ else
+ AC_MSG_RESULT([no])
+ AC_MSG_ERROR([pytest not found])
+ fi
+ fi
+fi
+
# If compiler will take -Wl,--as-needed (or various platform-specific
# spellings thereof) then add that to LDFLAGS. This is much easier than
# trying to filter LIBS to the minimum for each executable.
diff --git a/meson.build b/meson.build
index 5122706477d..ba53483f551 100644
--- a/meson.build
+++ b/meson.build
@@ -1824,6 +1824,47 @@ endif
+###############################################################
+# Library: pytest
+###############################################################
+
+pytest_enabled = false
+pytest_version = ''
+pytest_cmd = ['pytest'] # dummy, overwritten when pytest is found
+# We also configure the same PYTHONPATH in the pytest settings in
+# pyproject.toml, but pytest versions below 8.4 only actually use that
+# value after plugin loading. On lower versions pytest will throw an error even
+# when just running 'pytest --version'. So we need to configure it here too.
+# This won't help people manually running pytest outside of meson/make, but we
+# expect those to use a recent enough version of pytest anyway (and if not they
+# can manually configure PYTHONPATH too).
+pytest_env = {'PYTHONPATH': meson.project_source_root() / 'src' / 'test' / 'pytest'}
+
+pytestopt = get_option('pytest')
+if not pytestopt.disabled()
+ pytest = find_program(get_option('PYTEST'), native: true, required: false)
+
+ if pytest.found()
+ pytest_enabled = true
+ pytest_version = run_command(pytest, '--version', env: pytest_env, check: false).stdout().strip().split(' ')[-1]
+ pytest_cmd = [pytest.full_path()]
+ else
+ # Try python -m pytest as a fallback
+ pytest_check = run_command(python, '-m', 'pytest', '--version', env: pytest_env, check: false)
+ if pytest_check.returncode() == 0
+ pytest_enabled = true
+ pytest_version = pytest_check.stdout().strip().split(' ')[-1]
+ pytest_cmd = [python.full_path(), '-m', 'pytest']
+ endif
+ endif
+
+ if not pytest_enabled and pytestopt.enabled()
+ error('pytest not found')
+ endif
+endif
+
+
+
###############################################################
# Library: zstd
###############################################################
@@ -3923,6 +3964,64 @@ foreach test_dir : tests
)
endforeach
install_suites += test_group
+ elif kind == 'pytest'
+ testwrap_pytest = testwrap_base
+ if not pytest_enabled
+ testwrap_pytest += ['--skip', 'pytest not enabled']
+ endif
+
+ test_command = pytest_cmd
+
+ test_command += [
+ '-c', meson.project_source_root() / 'pyproject.toml',
+ '--verbose',
+ '-p', 'pgtap', # enable our test reporter plugin
+ '-ra', # show skipped and xfailed tests too
+ ]
+
+ # Add temporary install, the build directory for non-installed binaries and
+ # also test/ for non-installed test binaries built separately.
+ env = test_env
+ env.prepend('PATH', temp_install_bindir, test_dir['bd'], test_dir['bd'] / 'test')
+ temp_install_datadir = '@0@@1@'.format(test_install_destdir, dir_prefix / dir_data)
+ env.set('share_contrib_dir', temp_install_datadir / 'contrib')
+ env.prepend('PYTHONPATH', pytest_env['PYTHONPATH'])
+
+ foreach name, value : t.get('env', {})
+ env.set(name, value)
+ endforeach
+
+ test_group = test_dir['name']
+ test_kwargs = {
+ 'protocol': 'tap',
+ 'suite': test_group,
+ 'timeout': 1000,
+ 'depends': test_deps + t.get('deps', []),
+ 'env': env,
+ } + t.get('test_kwargs', {})
+
+ foreach onetest : t['tests']
+ # Make test names prettier, remove pyt/ and .py
+ onetest_p = onetest
+ if onetest_p.startswith('pyt/')
+ onetest_p = onetest.split('pyt/')[1]
+ endif
+ if onetest_p.endswith('.py')
+ onetest_p = fs.stem(onetest_p)
+ endif
+
+ test(test_dir['name'] / onetest_p,
+ python,
+ kwargs: test_kwargs,
+ args: testwrap_pytest + [
+ '--testgroup', test_dir['name'],
+ '--testname', onetest_p,
+ '--', test_command,
+ test_dir['sd'] / onetest,
+ ],
+ )
+ endforeach
+ install_suites += test_group
else
error('unknown kind @0@ of test in @1@'.format(kind, test_dir['sd']))
endif
@@ -4096,6 +4195,7 @@ summary(
'bison': '@0@ @1@'.format(bison.full_path(), bison_version),
'dtrace': dtrace,
'flex': '@0@ @1@'.format(flex.full_path(), flex_version),
+ 'pytest': pytest_enabled ? ' '.join(pytest_cmd) + ' ' + pytest_version : not_found_dep,
},
section: 'Programs',
)
diff --git a/meson_options.txt b/meson_options.txt
index 6a793f3e479..cb4825c3575 100644
--- a/meson_options.txt
+++ b/meson_options.txt
@@ -41,7 +41,10 @@ option('cassert', type: 'boolean', value: false,
description: 'Enable assertion checks (for debugging)')
option('tap_tests', type: 'feature', value: 'auto',
- description: 'Enable TAP tests')
+ description: 'Enable (Perl-based) TAP tests')
+
+option('pytest', type: 'feature', value: 'auto',
+ description: 'Enable (Python-based) pytest suites')
option('injection_points', type: 'boolean', value: false,
description: 'Enable injection points')
@@ -195,6 +198,9 @@ option('PERL', type: 'string', value: 'perl',
option('PROVE', type: 'string', value: 'prove',
description: 'Path to prove binary')
+option('PYTEST', type: 'array', value: ['pytest', 'py.test'],
+ description: 'Path to pytest binary')
+
option('PYTHON', type: 'array', value: ['python3', 'python'],
description: 'Path to python binary')
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 00000000000..60abb4d0655
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,21 @@
+[project]
+name = "postgresql-hackers-tooling"
+version = "0.1.0"
+description = "Pytest infrastructure for PostgreSQL"
+requires-python = ">=3.6"
+dependencies = [
+ # pytest 7.0 was the last version which supported Python 3.6, but the BSDs
+ # have started putting 8.x into ports, so we support both. (pytest 8 can be
+ # used throughout once we drop support for Python 3.7.)
+ "pytest >= 7.0, < 10",
+
+ # Any other dependencies are effectively optional (added below). We import
+ # these libraries using pytest.importorskip(). So tests will be skipped if
+ # they are not available.
+]
+
+[tool.pytest.ini_options]
+minversion = "7.0"
+
+# Common test code can be found here.
+pythonpath = ["src/test/pytest"]
diff --git a/src/Makefile.global.in b/src/Makefile.global.in
index 947a2d79e29..572c24c3f55 100644
--- a/src/Makefile.global.in
+++ b/src/Makefile.global.in
@@ -211,6 +211,7 @@ enable_dtrace = @enable_dtrace@
enable_coverage = @enable_coverage@
enable_injection_points = @enable_injection_points@
enable_tap_tests = @enable_tap_tests@
+enable_pytest = @enable_pytest@
python_includespec = @python_includespec@
python_libdir = @python_libdir@
@@ -356,6 +357,7 @@ MSGFMT = @MSGFMT@
MSGFMT_FLAGS = @MSGFMT_FLAGS@
MSGMERGE = @MSGMERGE@
OPENSSL = @OPENSSL@
+PYTEST = @PYTEST@
PYTHON = @PYTHON@
TAR = @TAR@
XGETTEXT = @XGETTEXT@
@@ -510,6 +512,33 @@ prove_installcheck = @echo "TAP tests not enabled. Try configuring with --enable
prove_check = $(prove_installcheck)
endif
+ifeq ($(enable_pytest),yes)
+
+pytest_installcheck = @echo "Installcheck is not currently supported for pytest."
+
+# We also configure the same PYTHONPATH in the pytest settings in
+# pyproject.toml, but pytest versions below 8.4 only actually use that value
+# after plugin loading. So we need to configure it here too. This won't help
+# people manually running pytest outside of meson/make, but we expect those to
+# use a recent enough version of pytest anyway (and if not they can manually
+# configure PYTHONPATH too).
+define pytest_check
+echo "# +++ pytest check in $(subdir) +++" && \
+rm -rf '$(CURDIR)'/tmp_check && \
+$(MKDIR_P) '$(CURDIR)'/tmp_check && \
+cd $(srcdir) && \
+ TESTLOGDIR='$(CURDIR)/tmp_check/log' \
+ TESTDATADIR='$(CURDIR)/tmp_check' \
+ PYTHONPATH='$(abs_top_srcdir)/src/test/pytest:$$PYTHONPATH' \
+ $(with_temp_install) \
+ $(PYTEST) -c '$(abs_top_srcdir)/pyproject.toml' --verbose -ra ./pyt/
+endef
+
+else
+pytest_installcheck = @echo "pytest is not enabled. Try configuring with --enable-pytest"
+pytest_check = $(pytest_installcheck)
+endif
+
# Installation.
install_bin = @install_bin@
diff --git a/src/makefiles/meson.build b/src/makefiles/meson.build
index 77f7a729cc2..6815178b35e 100644
--- a/src/makefiles/meson.build
+++ b/src/makefiles/meson.build
@@ -56,6 +56,8 @@ pgxs_kv = {
'enable_nls': libintl.found() ? 'yes' : 'no',
'enable_injection_points': get_option('injection_points') ? 'yes' : 'no',
'enable_tap_tests': tap_tests_enabled ? 'yes' : 'no',
+ 'enable_pytest': pytest_enabled ? 'yes' : 'no',
+ 'PYTEST': pytest_enabled ? ' '.join(pytest_cmd) : '',
'enable_debug': get_option('debug') ? 'yes' : 'no',
'enable_coverage': 'no',
'enable_dtrace': dtrace.found() ? 'yes' : 'no',
diff --git a/src/test/Makefile b/src/test/Makefile
index 3eb0a06abb4..0be9771d71f 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -18,6 +18,7 @@ SUBDIRS = \
modules \
perl \
postmaster \
+ pytest \
recovery \
regress \
subscription
diff --git a/src/test/meson.build b/src/test/meson.build
index cd45cbf57fb..09175f0eaea 100644
--- a/src/test/meson.build
+++ b/src/test/meson.build
@@ -5,6 +5,7 @@ subdir('isolation')
subdir('authentication')
subdir('postmaster')
+subdir('pytest')
subdir('recovery')
subdir('subscription')
subdir('modules')
diff --git a/src/test/pytest/Makefile b/src/test/pytest/Makefile
new file mode 100644
index 00000000000..2bdca96ccbe
--- /dev/null
+++ b/src/test/pytest/Makefile
@@ -0,0 +1,20 @@
+#-------------------------------------------------------------------------
+#
+# Makefile for pytest
+#
+# Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/test/pytest/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/test/pytest
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+
+check:
+ $(pytest_check)
+
+clean distclean maintainer-clean:
+ rm -rf tmp_check
diff --git a/src/test/pytest/README b/src/test/pytest/README
new file mode 100644
index 00000000000..1333ed77b7e
--- /dev/null
+++ b/src/test/pytest/README
@@ -0,0 +1 @@
+TODO
diff --git a/src/test/pytest/meson.build b/src/test/pytest/meson.build
new file mode 100644
index 00000000000..b1f6061b307
--- /dev/null
+++ b/src/test/pytest/meson.build
@@ -0,0 +1,15 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+if not pytest_enabled
+ subdir_done()
+endif
+
+tests += {
+ 'name': 'pytest',
+ 'sd': meson.current_source_dir(),
+ 'bd': meson.current_build_dir(),
+ 'pytest': {
+ 'tests': [
+ ],
+ },
+}
diff --git a/src/test/pytest/pgtap.py b/src/test/pytest/pgtap.py
new file mode 100644
index 00000000000..2ae16b624d5
--- /dev/null
+++ b/src/test/pytest/pgtap.py
@@ -0,0 +1,197 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import os
+import sys
+
+import pytest
+
+#
+# Helpers
+#
+
+
+class TAP:
+ """
+ A basic API for reporting via the TAP 12 protocol.
+
+ https://testanything.org/tap-specification.html
+ """
+
+ def __init__(self):
+ self.count = 0
+
+ def expect(self, num: int):
+ self.print(f"1..{num}")
+
+ def print(self, *args):
+ print(*args, file=sys.__stdout__)
+
+ def ok(self, name: str):
+ self.count += 1
+ self.print("ok", self.count, "-", name)
+
+ def skip(self, name: str, reason: str):
+ self.count += 1
+ self.print("ok", self.count, "-", name, "# skip", reason)
+
+ def fail(self, name: str, details: str):
+ self.count += 1
+ self.print("not ok", self.count, "-", name)
+
+ # mtest has some odd behavior around TAP tests where it won't print
+ # diagnostics on failure if they're part of the stdout stream, so we
+ # might as well just dump the details directly to stderr instead.
+ print(details, file=sys.__stderr__)
+
+
+tap = TAP()
+
+
+class TestNotes:
+ """
+ Annotations for a single test. The existing pytest hooks keep interesting
+ information somewhat separated across the different stages
+ (setup/test/teardown), so this class is used to correlate them.
+ """
+
+ skipped = False
+ skip_reason = None
+
+ failed = False
+ details = ""
+
+
+# Register a custom key in the stash dictionary for keeping our TestNotes.
+notes_key = pytest.StashKey[TestNotes]()
+
+
+#
+# Hook Implementations
+#
+
+
[email protected](tryfirst=True)
+def pytest_configure(config):
+ """
+ Hijacks the standard streams as soon as possible during pytest startup. The
+ pytest-formatted output gets logged to file instead, and we'll use the
+ original sys.__stdout__/__stderr__ streams for the TAP protocol.
+ """
+ logdir = os.getenv("TESTLOGDIR")
+ if not logdir:
+ raise RuntimeError("pgtap requires the TESTLOGDIR envvar to be set")
+
+ os.makedirs(logdir)
+ logpath = os.path.join(logdir, "pytest.log")
+ sys.stdout = sys.stderr = open(logpath, "a", buffering=1)
+
+
[email protected](trylast=True)
+def pytest_sessionfinish(session, exitstatus):
+ """
+ Suppresses nonzero exit codes due to failed tests. (In that case, we want
+ Meson to report a failure count, not a generic ERROR.)
+ """
+ if exitstatus == pytest.ExitCode.TESTS_FAILED:
+ session.exitstatus = pytest.ExitCode.OK
+
+
[email protected]
+def pytest_collectreport(report):
+ # Include collection failures directly in Meson error output.
+ if report.failed:
+ print(report.longreprtext, file=sys.__stderr__)
+
+
[email protected]
+def pytest_internalerror(excrepr, excinfo):
+ # Include internal errors directly in Meson error output.
+ print(excrepr, file=sys.__stderr__)
+
+
+#
+# Hook Wrappers
+#
+# In pytest parlance, a "wrapper" for a hook can inspect and optionally modify
+# existing hooks' behavior, but it does not replace the hook chain. This is done
+# through a generator-style API which chains the hooks together (see the use of
+# `yield`).
+#
+
+
[email protected](hookwrapper=True)
+def pytest_collection(session):
+ """Reports the number of gathered tests after collection is finished."""
+ res = yield
+ tap.expect(session.testscollected)
+ return res
+
+
[email protected](hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+ """
+ Annotates a test item with our TestNotes and grabs relevant information for
+ reporting.
+
+ This is called multiple times per test, so it's not correct to print the TAP
+ result here. (A test and its teardown stage can both fail, and we want to
+ see the details for both.) We instead combine all the information for use by
+ our pytest_runtest_protocol wrapper later on.
+ """
+ res = yield
+
+ if notes_key not in item.stash:
+ item.stash[notes_key] = TestNotes()
+ notes = item.stash[notes_key]
+
+ report = res.get_result()
+ if report.passed:
+ pass # no annotation needed
+
+ elif report.skipped:
+ notes.skipped = True
+ _, _, notes.skip_reason = report.longrepr
+
+ elif report.failed:
+ notes.failed = True
+
+ if not notes.details:
+ notes.details += "{:_^72}\n\n".format(f" {report.head_line} ")
+
+ if report.when in ("setup", "teardown"):
+ notes.details += "\n{:_^72}\n\n".format(
+ f" Error during {report.when} of {report.head_line} "
+ )
+
+ notes.details += report.longreprtext + "\n"
+
+ # Include captured stdout/stderr/log in failure output
+ for section_name, section_content in report.sections:
+ if section_content.strip():
+ notes.details += "\n{:-^72}\n".format(f" {section_name} ")
+ notes.details += section_content + "\n"
+
+ else:
+ raise RuntimeError("pytest_runtest_makereport received unknown test status")
+
+ return res
+
+
[email protected](hookwrapper=True)
+def pytest_runtest_protocol(item, nextitem):
+ """
+ Reports the TAP result for this test item using our gathered TestNotes.
+ """
+ res = yield
+
+ assert notes_key in item.stash, "pgtap didn't annotate a test item?"
+ notes = item.stash[notes_key]
+
+ if notes.failed:
+ tap.fail(item.nodeid, notes.details)
+ elif notes.skipped:
+ tap.skip(item.nodeid, notes.skip_reason)
+ else:
+ tap.ok(item.nodeid)
+
+ return res
diff --git a/src/tools/testwrap b/src/tools/testwrap
index e91296ecd15..346f86b8ea3 100755
--- a/src/tools/testwrap
+++ b/src/tools/testwrap
@@ -42,7 +42,11 @@ open(os.path.join(testdir, 'test.start'), 'x')
env_dict = {**os.environ,
'TESTDATADIR': os.path.join(testdir, 'data'),
- 'TESTLOGDIR': os.path.join(testdir, 'log')}
+ 'TESTLOGDIR': os.path.join(testdir, 'log'),
+ # Prevent emitting terminal capability sequences that pollute the
+ # TAP output stream (i.e.\033[?1034h). This happens on OpenBSD with
+ # pytest for unknown reasons.
+ 'TERM': ''}
# The configuration time value of PG_TEST_EXTRA is supplied via argument
base-commit: d80b0225010fd407c784bbecde116a28198b6eab
--
2.53.0
[text/x-patch] v12-0002-Add-pytest-infrastructure-to-interact-with-Postg.patch (73.7K, 5-v12-0002-Add-pytest-infrastructure-to-interact-with-Postg.patch)
download | inline diff:
From 9cdfb3ee319fa0700ae1f65399d9ed41a0115cb7 Mon Sep 17 00:00:00 2001
From: Jelte Fennema-Nio <[email protected]>
Date: Tue, 16 Dec 2025 09:25:48 +0100
Subject: [PATCH v12 2/5] Add pytest infrastructure to interact with PostgreSQL
servers
This adds functionality to the pytest infrastructure that allows tests
to do common things with PostgreSQL servers like:
- creating
- starting
- stopping
- connecting
- running queries
- handling errors
The goal of this infrastructure is to be so easy to use that the actual
tests really only contain the logic to test the behaviour that the tests
are testing, as opposed to a bunch of boilerplate. Examples of this are:
Types get converted to their Python counter parts automatically. Errors
become actual Python exceptions. Results of queries that only return a
single row or cell are unpacked automatically, so you don't have to do
rows[0][0] if the query only returns a single cell.
The only new tests that are part of this commit are tests that cover
this testing infrastructure itself. It's debatable whether such tests
are useful long term, because any infrastructure that's unused by actual
tests should probably not exist. For now it seems good to test this
basic functionality though, both to make sure we don't break it before
committing actual tests that use it, and also as an example for people
writing new tests.
---
doc/src/sgml/regress.sgml | 66 ++-
pyproject.toml | 3 +
src/test/pytest/README | 154 ++++++-
src/test/pytest/libpq/__init__.py | 35 ++
src/test/pytest/libpq/_core.py | 488 ++++++++++++++++++++++
src/test/pytest/libpq/errors.py | 62 +++
src/test/pytest/meson.build | 4 +
src/test/pytest/pypg/__init__.py | 10 +
src/test/pytest/pypg/_env.py | 72 ++++
src/test/pytest/pypg/fixtures.py | 335 +++++++++++++++
src/test/pytest/pypg/server.py | 476 +++++++++++++++++++++
src/test/pytest/pypg/util.py | 42 ++
src/test/pytest/pyt/conftest.py | 1 +
src/test/pytest/pyt/test_errors.py | 34 ++
src/test/pytest/pyt/test_libpq.py | 35 ++
src/test/pytest/pyt/test_multi_server.py | 46 ++
src/test/pytest/pyt/test_query_helpers.py | 347 +++++++++++++++
17 files changed, 2208 insertions(+), 2 deletions(-)
create mode 100644 src/test/pytest/libpq/__init__.py
create mode 100644 src/test/pytest/libpq/_core.py
create mode 100644 src/test/pytest/libpq/errors.py
create mode 100644 src/test/pytest/pypg/__init__.py
create mode 100644 src/test/pytest/pypg/_env.py
create mode 100644 src/test/pytest/pypg/fixtures.py
create mode 100644 src/test/pytest/pypg/server.py
create mode 100644 src/test/pytest/pypg/util.py
create mode 100644 src/test/pytest/pyt/conftest.py
create mode 100644 src/test/pytest/pyt/test_errors.py
create mode 100644 src/test/pytest/pyt/test_libpq.py
create mode 100644 src/test/pytest/pyt/test_multi_server.py
create mode 100644 src/test/pytest/pyt/test_query_helpers.py
diff --git a/doc/src/sgml/regress.sgml b/doc/src/sgml/regress.sgml
index d80dd46c5fd..2d85edacec7 100644
--- a/doc/src/sgml/regress.sgml
+++ b/doc/src/sgml/regress.sgml
@@ -840,7 +840,7 @@ float4:out:.*-.*-cygwin.*=float4-misrounded-input.out
</sect1>
<sect1 id="regress-tap">
- <title>TAP Tests</title>
+ <title>Perl TAP Tests</title>
<para>
Various tests, particularly the client program tests
@@ -929,6 +929,70 @@ PG_TEST_NOCLEAN=1 make -C src/bin/pg_dump check
</sect1>
+ <sect1 id="regress-pytest">
+ <title>Pytest Tests</title>
+
+ <para>
+ Tests in <filename>pyt</filename> directories use the Python
+ <application>pytest</application> framework. These tests provide a
+ convenient way to test libpq client functionality and scenarios requiring
+ multiple PostgreSQL server instances.
+ </para>
+
+ <para>
+ The pytest tests require <productname>PostgreSQL</productname> to be
+ configured with the option <option>--enable-pytest</option> (or
+ <option>-Dpytest=enabled</option> for Meson builds). You also need
+ <application>pytest</application> installed. You can either install it
+ system-wide, or create a virtual environment in the source directory:
+<programlisting>
+python -m venv .venv
+source .venv/bin/activate
+pip install .
+</programlisting>
+ Alternatively, if you have <application>uv</application> installed:
+<programlisting>
+uv sync
+source .venv/bin/activate
+</programlisting>
+ Remember to activate the virtual environment before running
+ <command>configure</command> or <command>meson setup</command>.
+ </para>
+
+ <para>
+ With Meson builds, you can run the pytest tests using:
+<programlisting>
+meson test --suite pytest
+</programlisting>
+ With autoconf-based builds, you can run them from the
+ <filename>src/test/pytest</filename> directory using:
+<programlisting>
+make check
+</programlisting>
+ </para>
+
+ <para>
+ You can also run specific test files directly using pytest:
+<programlisting>
+pytest src/test/pytest/pyt/test_libpq.py
+pytest -k "test_connstr"
+</programlisting>
+ </para>
+
+ <para>
+ Many operations in the test suites use a 180-second timeout, which on slow
+ hosts may lead to load-induced timeouts. Setting the environment variable
+ <varname>PG_TEST_TIMEOUT_DEFAULT</varname> to a higher number will change
+ the default to avoid this.
+ </para>
+
+ <para>
+ For more information on writing pytest tests, see the
+ <filename>src/test/pytest/README</filename> file.
+ </para>
+
+ </sect1>
+
<sect1 id="regress-coverage">
<title>Test Coverage Examination</title>
diff --git a/pyproject.toml b/pyproject.toml
index 60abb4d0655..4628d2274e0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -19,3 +19,6 @@ minversion = "7.0"
# Common test code can be found here.
pythonpath = ["src/test/pytest"]
+
+# Load the shared fixtures plugin
+addopts = ["-p", "pypg.fixtures"]
diff --git a/src/test/pytest/README b/src/test/pytest/README
index 1333ed77b7e..bb75e56a25d 100644
--- a/src/test/pytest/README
+++ b/src/test/pytest/README
@@ -1 +1,153 @@
-TODO
+src/test/pytest/README
+
+Pytest-based tests
+==================
+
+This directory contains infrastructure for Python-based tests using pytest,
+along with some core tests for the pytest infrastructure itself. The framework
+provides fixtures for managing PostgreSQL server instances and connecting to
+them via libpq.
+
+
+Running the tests
+=================
+
+NOTE: You must have given the --enable-pytest argument to configure (or
+-Dpytest=enabled for Meson builds). You also need to have pytest installed.
+
+If you don't have pytest installed system-wide, you can create a virtual
+environment:
+
+ python3 -m venv .venv
+ source .venv/bin/activate # On Windows: .venv\Scripts\activate
+ pip install . # Installs pytest and other dependencies
+
+Or using uv (https://docs.astral.sh/uv/):
+
+ uv sync
+ source .venv/bin/activate # On Windows: .venv\Scripts\activate
+
+Remember to activate the virtual environment before running configure/meson
+setup.
+
+With Meson builds, you can run:
+ meson test --suite pytest
+
+With autoconf based builds, you can run:
+ make check
+or
+ make installcheck
+
+You can run specific test files and/or use pytest's -k option to select tests:
+ pytest src/test/pytest/pyt/test_libpq.py
+ pytest -k "test_connstr"
+
+
+Directory structure
+===================
+
+pypg/
+ Python library providing common functions and pytest fixtures that can be
+ used in tests.
+
+libpq/
+ A simple but user-friendly python wrapper around libpq
+
+pyt/
+ Tests for the pytest infrastructure itself
+
+pgtap.py
+ A pytest plugin to output results in TAP format
+
+
+Writing tests
+=============
+
+Tests use pytest fixtures to manage server instances and connections. The
+most commonly used fixtures are:
+
+pg
+ A PostgresServer instance configured for the current test. Use this for
+ creating test users/databases or modifying server configuration. Changes
+ are automatically rolled back after the test.
+
+conn
+ A connected PGconn instance to the test server. Automatically cleaned up
+ after the test.
+
+connect
+ A function to create additional connections with custom options.
+
+create_pg
+ A factory function to create additional PostgreSQL servers within a test.
+ Servers are automatically cleaned up at the end of the test. Useful for
+ testing scenarios that require multiple independent servers.
+
+create_pg_module
+ Like create_pg, but servers persist for the entire test module. Use this
+ when multiple tests in a module can share the same servers, which is
+ faster than creating new servers for each test.
+
+
+Example test:
+
+ def test_simple_query(conn):
+ result = conn.sql("SELECT 1 + 1")
+ assert result == 2
+
+ def test_with_user(pg):
+ users = pg.create_users("test")
+ with pg.reloading() as s:
+ s.hba.prepend(["local", "all", users["test"], "trust"])
+
+ conn = pg.connect(user=users["test"])
+ assert conn.sql("SELECT current_user") == users["test"]
+
+ def test_multiple_servers(create_pg):
+ node1 = create_pg("primary")
+ node2 = create_pg("secondary")
+
+ conn1 = node1.connect()
+ conn2 = node2.connect()
+
+ # Each server is independent
+ assert node1.port != node2.port
+
+
+Server configuration
+====================
+
+Tests can temporarily modify server configuration using context managers:
+
+ with pg.reloading() as s:
+ s.conf.set(log_connections="on")
+ s.hba.prepend("local all all trust")
+ # Server is reloaded here
+ # After the test finished the original configuration is restored and
+ # the server is reloaded again
+
+Use pg.restarting() instead if the configuration change requires a restart.
+
+
+Timeouts
+========
+
+Tests inherit the PG_TEST_TIMEOUT_DEFAULT environment variable (defaulting
+to 180 seconds). The remaining_timeout fixture provides a function that
+returns how much time remains for the current test.
+
+
+Environment variables
+=====================
+
+PG_TEST_TIMEOUT_DEFAULT
+ Per-test timeout in seconds (default: 180)
+
+PG_CONFIG
+ Path to pg_config (default: uses PATH)
+
+TESTDATADIR
+ Directory for test data (default: pytest temp directory)
+
+PG_TEST_EXTRA
+ Space-separated list of optional test categories to run (e.g., "ssl")
diff --git a/src/test/pytest/libpq/__init__.py b/src/test/pytest/libpq/__init__.py
new file mode 100644
index 00000000000..6a71ebbe43f
--- /dev/null
+++ b/src/test/pytest/libpq/__init__.py
@@ -0,0 +1,35 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+libpq testing utilities - ctypes bindings and helpers for PostgreSQL's libpq library.
+
+This module provides Python wrappers around libpq for use in pytest tests.
+"""
+
+from . import errors
+from .errors import LibpqError
+from ._core import (
+ ConnectionStatus,
+ DiagField,
+ ExecStatus,
+ PGconn,
+ PGresult,
+ connect,
+ connstr,
+ load_libpq_handle,
+ register_type_info,
+)
+
+__all__ = [
+ "errors",
+ "LibpqError",
+ "ConnectionStatus",
+ "DiagField",
+ "ExecStatus",
+ "PGconn",
+ "PGresult",
+ "connect",
+ "connstr",
+ "load_libpq_handle",
+ "register_type_info",
+]
diff --git a/src/test/pytest/libpq/_core.py b/src/test/pytest/libpq/_core.py
new file mode 100644
index 00000000000..1c059b9b446
--- /dev/null
+++ b/src/test/pytest/libpq/_core.py
@@ -0,0 +1,488 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Core libpq functionality - ctypes bindings and connection handling.
+"""
+
+import contextlib
+import ctypes
+import datetime
+import decimal
+import enum
+import json
+import platform
+import os
+import uuid
+from typing import Any, Callable, Dict, Optional
+
+from .errors import LibpqError
+
+
+# PG_DIAG field identifiers from postgres_ext.h
+class DiagField(enum.IntEnum):
+ SEVERITY = ord("S")
+ SEVERITY_NONLOCALIZED = ord("V")
+ SQLSTATE = ord("C")
+ MESSAGE_PRIMARY = ord("M")
+ MESSAGE_DETAIL = ord("D")
+ MESSAGE_HINT = ord("H")
+ STATEMENT_POSITION = ord("P")
+ INTERNAL_POSITION = ord("p")
+ INTERNAL_QUERY = ord("q")
+ CONTEXT = ord("W")
+ SCHEMA_NAME = ord("s")
+ TABLE_NAME = ord("t")
+ COLUMN_NAME = ord("c")
+ DATATYPE_NAME = ord("d")
+ CONSTRAINT_NAME = ord("n")
+ SOURCE_FILE = ord("F")
+ SOURCE_LINE = ord("L")
+ SOURCE_FUNCTION = ord("R")
+
+
+class ConnectionStatus(enum.IntEnum):
+ """PostgreSQL connection status codes from libpq."""
+
+ CONNECTION_OK = 0
+ CONNECTION_BAD = 1
+
+
+class ExecStatus(enum.IntEnum):
+ """PostgreSQL result status codes from PQresultStatus."""
+
+ PGRES_EMPTY_QUERY = 0
+ PGRES_COMMAND_OK = 1
+ PGRES_TUPLES_OK = 2
+ PGRES_COPY_OUT = 3
+ PGRES_COPY_IN = 4
+ PGRES_BAD_RESPONSE = 5
+ PGRES_NONFATAL_ERROR = 6
+ PGRES_FATAL_ERROR = 7
+ PGRES_COPY_BOTH = 8
+ PGRES_SINGLE_TUPLE = 9
+ PGRES_PIPELINE_SYNC = 10
+ PGRES_PIPELINE_ABORTED = 11
+
+
+class _PGconn(ctypes.Structure):
+ pass
+
+
+class _PGresult(ctypes.Structure):
+ pass
+
+
+_PGconn_p = ctypes.POINTER(_PGconn)
+_PGresult_p = ctypes.POINTER(_PGresult)
+
+
+def load_libpq_handle(libdir, bindir):
+ """
+ Loads a ctypes handle for libpq. Some common function prototypes are
+ initialized for general use.
+ """
+ system = platform.system()
+
+ if system in ("Linux", "FreeBSD", "NetBSD", "OpenBSD"):
+ name = "libpq.so.5"
+ elif system == "Darwin":
+ name = "libpq.5.dylib"
+ elif system == "Windows":
+ name = "libpq.dll"
+ else:
+ assert False, f"the libpq fixture must be updated for {system}"
+
+ if system == "Windows":
+ # On Windows, libpq.dll is confusingly in bindir, not libdir. And we
+ # need to add this directory the the search path.
+ libpq_path = os.path.join(bindir, name)
+ lib = ctypes.CDLL(libpq_path)
+ else:
+ libpq_path = os.path.join(libdir, name)
+ lib = ctypes.CDLL(libpq_path)
+
+ #
+ # Function Prototypes
+ #
+
+ lib.PQconnectdb.restype = _PGconn_p
+ lib.PQconnectdb.argtypes = [ctypes.c_char_p]
+
+ lib.PQstatus.restype = ctypes.c_int
+ lib.PQstatus.argtypes = [_PGconn_p]
+
+ lib.PQexec.restype = _PGresult_p
+ lib.PQexec.argtypes = [_PGconn_p, ctypes.c_char_p]
+
+ lib.PQresultStatus.restype = ctypes.c_int
+ lib.PQresultStatus.argtypes = [_PGresult_p]
+
+ lib.PQclear.restype = None
+ lib.PQclear.argtypes = [_PGresult_p]
+
+ lib.PQerrorMessage.restype = ctypes.c_char_p
+ lib.PQerrorMessage.argtypes = [_PGconn_p]
+
+ lib.PQfinish.restype = None
+ lib.PQfinish.argtypes = [_PGconn_p]
+
+ lib.PQresultErrorMessage.restype = ctypes.c_char_p
+ lib.PQresultErrorMessage.argtypes = [_PGresult_p]
+
+ lib.PQntuples.restype = ctypes.c_int
+ lib.PQntuples.argtypes = [_PGresult_p]
+
+ lib.PQnfields.restype = ctypes.c_int
+ lib.PQnfields.argtypes = [_PGresult_p]
+
+ lib.PQgetvalue.restype = ctypes.c_char_p
+ lib.PQgetvalue.argtypes = [_PGresult_p, ctypes.c_int, ctypes.c_int]
+
+ lib.PQgetisnull.restype = ctypes.c_int
+ lib.PQgetisnull.argtypes = [_PGresult_p, ctypes.c_int, ctypes.c_int]
+
+ lib.PQftype.restype = ctypes.c_uint
+ lib.PQftype.argtypes = [_PGresult_p, ctypes.c_int]
+
+ lib.PQresultErrorField.restype = ctypes.c_char_p
+ lib.PQresultErrorField.argtypes = [_PGresult_p, ctypes.c_int]
+
+ return lib
+
+
+# PostgreSQL type OIDs and conversion system
+# Type registry - maps OID to converter function
+_type_converters: Dict[int, Callable[[str], Any]] = {}
+_array_to_elem_map: Dict[int, int] = {}
+
+
+def register_type_info(
+ name: str, oid: int, array_oid: int, converter: Callable[[str], Any]
+):
+ """
+ Register a PostgreSQL type with its OID, array OID, and conversion function.
+
+ Usage:
+ register_type_info("bool", 16, 1000, lambda v: v == "t")
+ """
+ _type_converters[oid] = converter
+ if array_oid is not None:
+ _array_to_elem_map[array_oid] = oid
+
+
+def _parse_array(value: str, elem_oid: int):
+ """Parse PostgreSQL array syntax into nested Python lists."""
+ stack: list[list] = []
+ current_element: list[str] = []
+ in_quotes = False
+ was_quoted = False
+ pos = 0
+
+ while pos < len(value):
+ char = value[pos]
+
+ if in_quotes:
+ if char == "\\":
+ next_char = value[pos + 1]
+ if next_char not in '"\\':
+ raise NotImplementedError('Only \\" and \\\\ escapes are supported')
+ current_element.append(next_char)
+ pos += 2
+ continue
+ elif char == '"':
+ in_quotes = False
+ else:
+ current_element.append(char)
+ elif char == '"':
+ in_quotes = True
+ was_quoted = True
+ elif char == "{":
+ stack.append([])
+ elif char in ",}":
+ if current_element or was_quoted:
+ elem = "".join(current_element)
+ if not was_quoted and elem == "NULL":
+ stack[-1].append(None)
+ else:
+ stack[-1].append(_convert_pg_value(elem, elem_oid))
+ current_element = []
+ was_quoted = False
+ if char == "}":
+ completed = stack.pop()
+ if not stack:
+ return completed
+ stack[-1].append(completed)
+ elif char != " ":
+ current_element.append(char)
+ pos += 1
+
+ raise ValueError(f"Malformed array literal: {value}")
+
+
+# Register standard PostgreSQL types that we'll likely encounter in tests
+register_type_info("bool", 16, 1000, lambda v: v == "t")
+register_type_info("int2", 21, 1005, int)
+register_type_info("int4", 23, 1007, int)
+register_type_info("int8", 20, 1016, int)
+register_type_info("float4", 700, 1021, float)
+register_type_info("float8", 701, 1022, float)
+register_type_info("numeric", 1700, 1231, decimal.Decimal)
+register_type_info("text", 25, 1009, str)
+register_type_info("varchar", 1043, 1015, str)
+register_type_info("date", 1082, 1182, datetime.date.fromisoformat)
+register_type_info("time", 1083, 1183, datetime.time.fromisoformat)
+register_type_info("timestamp", 1114, 1115, datetime.datetime.fromisoformat)
+register_type_info("timestamptz", 1184, 1185, datetime.datetime.fromisoformat)
+register_type_info("uuid", 2950, 2951, uuid.UUID)
+register_type_info("json", 114, 199, json.loads)
+register_type_info("jsonb", 3802, 3807, json.loads)
+
+
+def _convert_pg_value(value: str, type_oid: int) -> Any:
+ """
+ Convert PostgreSQL string value to appropriate Python type based on OID.
+ Uses the registered type converters from register_type_info().
+ """
+ # Check if it's an array type
+ if type_oid in _array_to_elem_map:
+ elem_oid = _array_to_elem_map[type_oid]
+ return _parse_array(value, elem_oid)
+
+ # Use registered converter if available
+ converter = _type_converters.get(type_oid)
+ if converter:
+ return converter(value)
+
+ # Unknown types - return as string
+ return value
+
+
+def simplify_query_results(results) -> Any:
+ """
+ Simplify the results of a query so that the caller doesn't have to unpack
+ lists and tuples of length 1.
+ """
+ if len(results) == 1:
+ row = results[0]
+ if len(row) == 1:
+ # If there's only a single cell, just return the value
+ return row[0]
+ # If there's only a single row, just return that row
+ return row
+
+ if len(results) != 0 and len(results[0]) == 1:
+ # If there's only a single column, return an array of values
+ return [row[0] for row in results]
+
+ # if there are multiple rows and columns, return the results as is
+ return results
+
+
+class PGresult(contextlib.AbstractContextManager):
+ """Wraps a raw _PGresult_p with a more friendly interface."""
+
+ def __init__(self, lib: ctypes.CDLL, res: _PGresult_p):
+ self._lib = lib
+ self._res = res
+
+ def __exit__(self, *exc):
+ self._lib.PQclear(self._res)
+ self._res = None
+
+ def status(self) -> ExecStatus:
+ return ExecStatus(self._lib.PQresultStatus(self._res))
+
+ def error_message(self):
+ """Returns the error message associated with this result."""
+ msg = self._lib.PQresultErrorMessage(self._res)
+ return msg.decode() if msg else ""
+
+ def _get_error_field(self, field: DiagField) -> Optional[str]:
+ """Get an error field from the result using PQresultErrorField."""
+ val = self._lib.PQresultErrorField(self._res, int(field))
+ return val.decode() if val else None
+
+ def raise_error(self) -> None:
+ """
+ Raises LibpqError with diagnostic information from the result.
+ """
+ if not self._res:
+ raise LibpqError("query failed: out of memory or connection lost")
+
+ sqlstate = self._get_error_field(DiagField.SQLSTATE)
+ primary = self._get_error_field(DiagField.MESSAGE_PRIMARY)
+ detail = self._get_error_field(DiagField.MESSAGE_DETAIL)
+ hint = self._get_error_field(DiagField.MESSAGE_HINT)
+ severity = self._get_error_field(DiagField.SEVERITY)
+ schema_name = self._get_error_field(DiagField.SCHEMA_NAME)
+ table_name = self._get_error_field(DiagField.TABLE_NAME)
+ column_name = self._get_error_field(DiagField.COLUMN_NAME)
+ datatype_name = self._get_error_field(DiagField.DATATYPE_NAME)
+ constraint_name = self._get_error_field(DiagField.CONSTRAINT_NAME)
+ context = self._get_error_field(DiagField.CONTEXT)
+
+ position_str = self._get_error_field(DiagField.STATEMENT_POSITION)
+ position = int(position_str) if position_str else None
+
+ raise LibpqError(
+ primary or self.error_message(),
+ sqlstate=sqlstate,
+ severity=severity,
+ primary=primary,
+ detail=detail,
+ hint=hint,
+ schema_name=schema_name,
+ table_name=table_name,
+ column_name=column_name,
+ datatype_name=datatype_name,
+ constraint_name=constraint_name,
+ position=position,
+ context=context,
+ )
+
+ def fetch_all(self):
+ """
+ Fetch all rows and convert to Python types.
+ Returns a list of tuples, with values converted based on their PostgreSQL type.
+ """
+ nrows = self._lib.PQntuples(self._res)
+ ncols = self._lib.PQnfields(self._res)
+
+ # Get type OIDs for each column
+ type_oids = [self._lib.PQftype(self._res, col) for col in range(ncols)]
+
+ results = []
+ for row in range(nrows):
+ row_data = []
+ for col in range(ncols):
+ if self._lib.PQgetisnull(self._res, row, col):
+ row_data.append(None)
+ else:
+ value = self._lib.PQgetvalue(self._res, row, col).decode()
+ row_data.append(_convert_pg_value(value, type_oids[col]))
+ results.append(tuple(row_data))
+
+ return results
+
+
+class PGconn(contextlib.AbstractContextManager):
+ """
+ Wraps a raw _PGconn_p with a more friendly interface. This is just a
+ stub; it's expected to grow.
+ """
+
+ def __init__(
+ self,
+ lib: ctypes.CDLL,
+ handle: _PGconn_p,
+ stack: contextlib.ExitStack,
+ ):
+ self._lib = lib
+ self._handle = handle
+ self._stack = stack
+
+ def __exit__(self, *exc):
+ self._lib.PQfinish(self._handle)
+ self._handle = None
+
+ def exec(self, query: str):
+ """
+ Executes a query via PQexec() and returns a PGresult.
+ """
+ res = self._lib.PQexec(self._handle, query.encode())
+ return self._stack.enter_context(PGresult(self._lib, res))
+
+ def sql(self, query: str):
+ """
+ Executes a query and raises an exception if it fails.
+ Returns the query results with automatic type conversion and simplification.
+ For commands that don't return data (INSERT, UPDATE, etc.), returns None.
+
+ Examples:
+ - SELECT 1 -> 1
+ - SELECT 1, 2 -> (1, 2)
+ - SELECT * FROM generate_series(1, 3) -> [1, 2, 3]
+ - SELECT * FROM (VALUES (1, 'a'), (2, 'b')) t -> [(1, 'a'), (2, 'b')]
+ - CREATE TABLE ... -> None
+ - INSERT INTO ... -> None
+ """
+ res = self.exec(query)
+ status = res.status()
+
+ if status == ExecStatus.PGRES_FATAL_ERROR:
+ res.raise_error()
+ elif status == ExecStatus.PGRES_COMMAND_OK:
+ return None
+ elif status == ExecStatus.PGRES_TUPLES_OK:
+ results = res.fetch_all()
+ return simplify_query_results(results)
+ else:
+ res.raise_error()
+
+
+def connstr(opts: Dict[str, Any]) -> str:
+ """
+ Flattens the provided options into a libpq connection string. Values
+ are converted to str and quoted/escaped as necessary.
+ """
+ settings = []
+
+ for k, v in opts.items():
+ v = str(v)
+ if not v:
+ v = "''"
+ else:
+ v = v.replace("\\", "\\\\")
+ v = v.replace("'", "\\'")
+
+ if " " in v:
+ v = f"'{v}'"
+
+ settings.append(f"{k}={v}")
+
+ return " ".join(settings)
+
+
+def connect(
+ libpq_handle: ctypes.CDLL,
+ stack: contextlib.ExitStack,
+ remaining_timeout_fn: Callable[[], float],
+ **opts,
+) -> PGconn:
+ """
+ Connects to a server, using the given connection options, and
+ returns a PGconn object wrapping the connection handle. A
+ failure will raise LibpqError.
+
+ Connections honor PG_TEST_TIMEOUT_DEFAULT unless connect_timeout is
+ explicitly overridden in opts.
+
+ Args:
+ libpq_handle: ctypes.CDLL handle to libpq library
+ stack: ExitStack for managing connection cleanup
+ remaining_timeout_fn: Function that returns remaining timeout in seconds
+ **opts: Connection options (host, port, dbname, etc.)
+
+ Returns:
+ PGconn: Connected database connection
+
+ Raises:
+ LibpqError: If connection fails
+ """
+
+ if "connect_timeout" not in opts:
+ t = int(remaining_timeout_fn())
+ opts["connect_timeout"] = max(t, 1)
+
+ conn_p = libpq_handle.PQconnectdb(connstr(opts).encode())
+
+ # Check connection status before adding to stack
+ if libpq_handle.PQstatus(conn_p) != ConnectionStatus.CONNECTION_OK:
+ error_msg = libpq_handle.PQerrorMessage(conn_p).decode()
+ # Manually close the failed connection
+ libpq_handle.PQfinish(conn_p)
+ raise LibpqError(error_msg)
+
+ # Connection succeeded - add to stack for cleanup
+ conn = stack.enter_context(PGconn(libpq_handle, conn_p, stack=stack))
+ return conn
diff --git a/src/test/pytest/libpq/errors.py b/src/test/pytest/libpq/errors.py
new file mode 100644
index 00000000000..c665b663e22
--- /dev/null
+++ b/src/test/pytest/libpq/errors.py
@@ -0,0 +1,62 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Exception classes for libpq errors.
+"""
+
+from typing import Optional
+
+
+class LibpqError(RuntimeError):
+ """Exception for libpq errors with PostgreSQL diagnostic fields."""
+
+ sqlstate: Optional[str]
+ severity: Optional[str]
+ primary: Optional[str]
+ detail: Optional[str]
+ hint: Optional[str]
+ schema_name: Optional[str]
+ table_name: Optional[str]
+ column_name: Optional[str]
+ datatype_name: Optional[str]
+ constraint_name: Optional[str]
+ position: Optional[int]
+ context: Optional[str]
+
+ def __init__(
+ self,
+ message: str,
+ *,
+ sqlstate: Optional[str] = None,
+ severity: Optional[str] = None,
+ primary: Optional[str] = None,
+ detail: Optional[str] = None,
+ hint: Optional[str] = None,
+ schema_name: Optional[str] = None,
+ table_name: Optional[str] = None,
+ column_name: Optional[str] = None,
+ datatype_name: Optional[str] = None,
+ constraint_name: Optional[str] = None,
+ position: Optional[int] = None,
+ context: Optional[str] = None,
+ ):
+ super().__init__(message)
+ self.sqlstate = sqlstate
+ self.severity = severity
+ self.primary = primary
+ self.detail = detail
+ self.hint = hint
+ self.schema_name = schema_name
+ self.table_name = table_name
+ self.column_name = column_name
+ self.datatype_name = datatype_name
+ self.constraint_name = constraint_name
+ self.position = position
+ self.context = context
+
+ @property
+ def sqlstate_class(self) -> Optional[str]:
+ """Returns the 2-character SQLSTATE class."""
+ if self.sqlstate and len(self.sqlstate) >= 2:
+ return self.sqlstate[:2]
+ return None
diff --git a/src/test/pytest/meson.build b/src/test/pytest/meson.build
index b1f6061b307..b86be901e7c 100644
--- a/src/test/pytest/meson.build
+++ b/src/test/pytest/meson.build
@@ -10,6 +10,10 @@ tests += {
'bd': meson.current_build_dir(),
'pytest': {
'tests': [
+ 'pyt/test_errors.py',
+ 'pyt/test_libpq.py',
+ 'pyt/test_multi_server.py',
+ 'pyt/test_query_helpers.py',
],
},
}
diff --git a/src/test/pytest/pypg/__init__.py b/src/test/pytest/pypg/__init__.py
new file mode 100644
index 00000000000..4ee91289f70
--- /dev/null
+++ b/src/test/pytest/pypg/__init__.py
@@ -0,0 +1,10 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+from ._env import require_test_extras, skip_unless_test_extras
+from .server import PostgresServer
+
+__all__ = [
+ "require_test_extras",
+ "skip_unless_test_extras",
+ "PostgresServer",
+]
diff --git a/src/test/pytest/pypg/_env.py b/src/test/pytest/pypg/_env.py
new file mode 100644
index 00000000000..c4087be3212
--- /dev/null
+++ b/src/test/pytest/pypg/_env.py
@@ -0,0 +1,72 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import logging
+import os
+
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+def _test_extra_skip_reason(*keys: str) -> str:
+ return "requires {} to be set in PG_TEST_EXTRA".format(", ".join(keys))
+
+
+def _has_test_extra(key: str) -> bool:
+ """
+ Returns True if the PG_TEST_EXTRA environment variable contains the given
+ key.
+ """
+ extra = os.getenv("PG_TEST_EXTRA", "")
+ return key in extra.split()
+
+
+def require_test_extras(*keys: str):
+ """
+ A convenience annotation which will skip tests if all of the required keys
+ are not present in PG_TEST_EXTRA.
+
+ To skip a particular test function or class:
+
+ @pypg.require_test_extras("ldap")
+ def test_some_ldap_feature():
+ ...
+
+ To skip an entire module:
+
+ pytestmark = pypg.require_test_extra("ssl", "kerberos")
+ """
+ return pytest.mark.skipif(
+ not all([_has_test_extra(k) for k in keys]),
+ reason=_test_extra_skip_reason(*keys),
+ )
+
+
+def skip_unless_test_extras(*keys: str):
+ """
+ Skip the current test/fixture if any of the required keys are not present
+ in PG_TEST_EXTRA. Use this inside fixtures where decorators can't be used.
+
+ @pytest.fixture
+ def my_fixture():
+ skip_unless_test_extras("ldap")
+ ...
+ """
+ if not all([_has_test_extra(k) for k in keys]):
+ pytest.skip(_test_extra_skip_reason(*keys))
+
+
+def test_timeout_default() -> int:
+ """
+ Returns the value of the PG_TEST_TIMEOUT_DEFAULT environment variable, in
+ seconds, or 180 if one was not provided.
+ """
+ default = os.getenv("PG_TEST_TIMEOUT_DEFAULT", "")
+ if not default:
+ return 180
+
+ try:
+ return int(default)
+ except ValueError as v:
+ logger.warning("PG_TEST_TIMEOUT_DEFAULT could not be parsed: " + str(v))
+ return 180
diff --git a/src/test/pytest/pypg/fixtures.py b/src/test/pytest/pypg/fixtures.py
new file mode 100644
index 00000000000..8c0cb60daa5
--- /dev/null
+++ b/src/test/pytest/pypg/fixtures.py
@@ -0,0 +1,335 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import os
+import contextlib
+import pathlib
+import time
+from typing import List
+
+import pytest
+
+from ._env import test_timeout_default
+from .util import capture
+from .server import PostgresServer
+
+from libpq import load_libpq_handle, connect as libpq_connect
+
+
+# Stash key for tracking servers for log reporting.
+_servers_key = pytest.StashKey[List[PostgresServer]]()
+
+
+def _record_server_for_log_reporting(request, server):
+ """Record a server for log reporting on test failure."""
+ if _servers_key not in request.node.stash:
+ request.node.stash[_servers_key] = []
+ request.node.stash[_servers_key].append(server)
+
+
[email protected]
+def remaining_timeout():
+ """
+ This fixture provides a function that returns how much of the
+ PG_TEST_TIMEOUT_DEFAULT remains for the current test, in fractional seconds.
+ This value is never less than zero.
+
+ This fixture is per-test, so the deadline is also reset on a per-test basis.
+ """
+ now = time.monotonic()
+ deadline = now + test_timeout_default()
+
+ return lambda: max(deadline - time.monotonic(), 0)
+
+
[email protected](scope="module")
+def remaining_timeout_module():
+ """
+ Same as remaining_timeout, but the deadline is set once per module.
+
+ This fixture is per-module, which means it's generally only really useful
+ for configuring timeouts of operations that happen in the setup phase of
+ another module fixtures. If you use it in a test it would mean that each
+ subsequent test in the module gets a reduced timeout.
+ """
+ now = time.monotonic()
+ deadline = now + test_timeout_default()
+
+ return lambda: max(deadline - time.monotonic(), 0)
+
+
[email protected](scope="session")
+def libpq_handle(libdir, bindir):
+ """
+ Loads a ctypes handle for libpq. Some common function prototypes are
+ initialized for general use.
+ """
+ try:
+ return load_libpq_handle(libdir, bindir)
+ except OSError as e:
+ if "wrong ELF class" in str(e):
+ # This happens in CI when trying to lead a 32-bit libpq library
+ # with a 64-bit Python
+ pytest.skip("libpq architecture does not match Python interpreter")
+ raise
+
+
[email protected]
+def connect(libpq_handle, remaining_timeout):
+ """
+ Returns a function to connect to PostgreSQL via libpq.
+
+ The returned function accepts connection options as keyword arguments
+ (host, port, dbname, etc.) and returns a PGconn object. Connections
+ are automatically cleaned up at the end of the test.
+
+ Example:
+ conn = connect(host='localhost', port=5432, dbname='postgres')
+ result = conn.sql("SELECT 1")
+ """
+ with contextlib.ExitStack() as stack:
+
+ def _connect(**opts):
+ return libpq_connect(libpq_handle, stack, remaining_timeout, **opts)
+
+ yield _connect
+
+
[email protected](scope="session")
+def pg_config():
+ """
+ Returns the path to pg_config. Uses PG_CONFIG environment variable if set,
+ otherwise uses 'pg_config' from PATH.
+ """
+ return os.environ.get("PG_CONFIG", "pg_config")
+
+
[email protected](scope="session")
+def bindir(pg_config):
+ """
+ Returns the PostgreSQL bin directory using pg_config --bindir.
+ """
+ return pathlib.Path(capture(pg_config, "--bindir"))
+
+
[email protected](scope="session")
+def libdir(pg_config):
+ """
+ Returns the PostgreSQL lib directory using pg_config --libdir.
+ """
+ return pathlib.Path(capture(pg_config, "--libdir"))
+
+
[email protected](scope="session")
+def tmp_check(tmp_path_factory) -> pathlib.Path:
+ """
+ Returns the tmp_check directory that should be used for the tests. If
+ TESTDATADIR is provided, that will be used; otherwise a new temporary
+ directory is created in the pytest temp root.
+ """
+ d = os.getenv("TESTDATADIR")
+ if d:
+ d = pathlib.Path(d)
+ else:
+ d = tmp_path_factory.mktemp("tmp_check")
+
+ return d
+
+
[email protected](scope="session")
+def datadir(tmp_check):
+ """
+ Returns the data directory to use for the pg fixture.
+ """
+
+ return tmp_check / "pgdata"
+
+
[email protected](scope="session")
+def sockdir(tmp_path_factory):
+ """
+ Returns the directory name to use as the server's unix_socket_directories
+ setting. Local client connections use this as the PGHOST.
+
+ At the moment, this is always put under the pytest temp root.
+ """
+ return tmp_path_factory.mktemp("sockfiles")
+
+
[email protected](scope="session")
+def pg_server_global(bindir, datadir, sockdir, libpq_handle):
+ """
+ Starts a running Postgres server listening on localhost. The HBA initially
+ allows only local UNIX connections from the same user.
+
+ Returns a PostgresServer instance with methods for server management, configuration,
+ and creating test databases/users.
+ """
+ server = PostgresServer("default", bindir, datadir, sockdir, libpq_handle)
+
+ yield server
+
+ # Cleanup any test resources
+ server.cleanup()
+
+ # Stop the server
+ server.stop()
+
+
[email protected](scope="module")
+def pg_server_module(pg_server_global):
+ """
+ Module-scoped server context. Which can be useful so that certain settings
+ can be overriden at the module level through autouse fixtures. An example
+ of this is in the SSL tests.
+ """
+ with pg_server_global.subcontext() as s:
+ yield s
+
+
[email protected]
+def pg(request, pg_server_module, remaining_timeout):
+ """
+ Per-test server context. Use this fixture to make changes to the server
+ which will be rolled back at the end of the test (e.g., creating test
+ users/databases).
+
+ Also captures the PostgreSQL log position at test start so that any new
+ log entries can be included in the test report on failure.
+ """
+ with pg_server_module.start_new_test(remaining_timeout) as s:
+ _record_server_for_log_reporting(request, s)
+ yield s
+
+
[email protected]
+def conn(pg):
+ """
+ Returns a connected PGconn instance to the test PostgreSQL server.
+ The connection is automatically cleaned up at the end of the test.
+
+ Example:
+ def test_something(conn):
+ result = conn.sql("SELECT 1")
+ assert result == 1
+ """
+ return pg.connect()
+
+
[email protected]
+def create_pg(request, bindir, sockdir, libpq_handle, tmp_check, remaining_timeout):
+ """
+ Factory fixture to create additional PostgreSQL servers (per-test scope).
+
+ Returns a function that creates new PostgreSQL server instances.
+ Servers are automatically cleaned up at the end of the test.
+
+ Example:
+ def test_multiple_servers(create_pg):
+ node1 = create_pg()
+ node2 = create_pg()
+ node3 = create_pg()
+ """
+ servers = []
+
+ def _create(name=None, **kwargs):
+ if name is None:
+ count = len(servers) + 1
+ name = f"pg{count}"
+
+ datadir = tmp_check / f"pgdata_{name}"
+ server = PostgresServer(name, bindir, datadir, sockdir, libpq_handle, **kwargs)
+ server.set_timeout(remaining_timeout)
+ _record_server_for_log_reporting(request, server)
+ servers.append(server)
+ return server
+
+ yield _create
+
+ for server in servers:
+ server.cleanup()
+ server.stop()
+
+
[email protected](scope="module")
+def _module_scoped_servers():
+ """Session-scoped list to track servers created by create_pg_module."""
+ return []
+
+
[email protected](scope="module")
+def create_pg_module(
+ bindir,
+ sockdir,
+ libpq_handle,
+ tmp_check,
+ remaining_timeout_module,
+ _module_scoped_servers,
+):
+ """
+ Factory fixture to create additional PostgreSQL servers (module scope).
+
+ Like create_pg, but servers persist for the entire test module.
+ Use this when multiple tests in a module can share the same servers.
+
+ The timeout is automatically set on all servers at the start of each test
+ via the _set_module_server_timeouts autouse fixture.
+
+ Example:
+ @pytest.fixture(scope="module")
+ def shared_nodes(create_pg_module):
+ return [create_pg_module() for _ in range(3)]
+ """
+
+ def _create(name=None, **kwargs):
+ if name is None:
+ count = len(_module_scoped_servers) + 1
+ name = f"pg{count}"
+ datadir = tmp_check / f"pgdata_{name}"
+ server = PostgresServer(name, bindir, datadir, sockdir, libpq_handle, **kwargs)
+ server.set_timeout(remaining_timeout_module)
+ _module_scoped_servers.append(server)
+ return server
+
+ yield _create
+
+ for server in _module_scoped_servers:
+ server.cleanup()
+ server.stop()
+
+
[email protected](autouse=True)
+def _set_module_server_timeouts(request, _module_scoped_servers, remaining_timeout):
+ """Autouse fixture that sets timeout, enters subcontext, and records log positions for module-scoped servers."""
+ with contextlib.ExitStack() as stack:
+ for server in _module_scoped_servers:
+ stack.enter_context(server.start_new_test(remaining_timeout))
+ _record_server_for_log_reporting(request, server)
+ yield
+
+
[email protected](hookwrapper=True, trylast=True)
+def pytest_runtest_makereport(item, call):
+ """
+ Adds PostgreSQL server logs to the test report sections.
+ """
+ outcome = yield
+ report = outcome.get_result()
+
+ if report.when != "call":
+ return
+
+ if _servers_key not in item.stash:
+ return
+
+ servers = item.stash[_servers_key]
+ del item.stash[_servers_key]
+
+ include_name = len(servers) > 1
+
+ for server in servers:
+ content = server.log_content()
+ if content.strip():
+ section_title = "Postgres log"
+ if include_name:
+ section_title += f" ({server.name})"
+ report.sections.append((section_title, content))
diff --git a/src/test/pytest/pypg/server.py b/src/test/pytest/pypg/server.py
new file mode 100644
index 00000000000..66496328d2d
--- /dev/null
+++ b/src/test/pytest/pypg/server.py
@@ -0,0 +1,476 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import contextlib
+import os
+import pathlib
+import platform
+import re
+import shutil
+import socket
+import subprocess
+import tempfile
+from collections import namedtuple
+from typing import Callable, Optional
+
+from .util import run
+from libpq import PGconn, connect as libpq_connect
+
+
+class FileBackup(contextlib.AbstractContextManager):
+ """
+ A context manager which backs up a file's contents, restoring them on exit.
+ """
+
+ def __init__(self, file: pathlib.Path):
+ super().__init__()
+
+ self._file = file
+
+ def __enter__(self):
+ with tempfile.NamedTemporaryFile(
+ prefix=self._file.name, dir=self._file.parent, delete=False
+ ) as f:
+ self._backup = pathlib.Path(f.name)
+
+ shutil.copyfile(self._file, self._backup)
+
+ return self
+
+ def __exit__(self, *exc):
+ # Swap the backup and the original file, so that the modified contents
+ # can still be inspected in case of failure.
+ tmp = self._backup.parent / (self._backup.name + ".tmp")
+
+ shutil.copyfile(self._file, tmp)
+ shutil.copyfile(self._backup, self._file)
+ shutil.move(tmp, self._backup)
+
+
+class HBA(FileBackup):
+ """
+ Backs up a server's HBA configuration and provides means for temporarily
+ editing it.
+ """
+
+ def __init__(self, datadir: pathlib.Path):
+ super().__init__(datadir / "pg_hba.conf")
+
+ def prepend(self, *lines):
+ """
+ Temporarily prepends lines to the server's pg_hba.conf.
+
+ As sugar for aligning HBA columns in the tests, each line can be either
+ a string or a list of strings. List elements will be joined by single
+ spaces before they are written to file.
+ """
+ with open(self._file, "r") as f:
+ prior_data = f.read()
+
+ with open(self._file, "w") as f:
+ for line in lines:
+ if isinstance(line, list):
+ print(*line, file=f)
+ else:
+ print(line, file=f)
+
+ f.write(prior_data)
+
+
+class Config(FileBackup):
+ """
+ Backs up a server's postgresql.conf and provides means for temporarily
+ editing it.
+ """
+
+ def __init__(self, datadir: pathlib.Path):
+ super().__init__(datadir / "postgresql.conf")
+
+ def set(self, **gucs):
+ """
+ Temporarily appends GUC settings to the server's postgresql.conf.
+ """
+
+ with open(self._file, "a") as f:
+ print(file=f)
+
+ for n, v in gucs.items():
+ v = str(v)
+
+ # Quote and escape the value for postgresql.conf single-quoted
+ # strings. This is doing the reversee of DeescapeQuotedString.
+ v = v.replace("\\", "\\\\")
+ v = v.replace("'", "''")
+ v = v.replace("\n", "\\n")
+ v = v.replace("\r", "\\r")
+ v = v.replace("\t", "\\t")
+ v = v.replace("\b", "\\b")
+ v = v.replace("\f", "\\f")
+ v = "'{}'".format(v)
+
+ print(n, "=", v, file=f)
+
+
+Backup = namedtuple("Backup", "conf, hba")
+
+
+class PostgresServer:
+ """
+ Represents a running PostgreSQL server instance with management utilities.
+ Provides methods for configuration, user/database creation, and server control.
+ """
+
+ def __init__(
+ self,
+ name,
+ bindir,
+ datadir,
+ sockdir,
+ libpq_handle,
+ *,
+ hostaddr: Optional[str] = None,
+ port: Optional[int] = None,
+ ):
+ """
+ Initialize and start a PostgreSQL server instance.
+
+ Args:
+ name: The name of this server instance (for logging purposes)
+ bindir: Path to PostgreSQL bin directory
+ datadir: Path to data directory for this server
+ sockdir: Path to directory for Unix sockets
+ libpq_handle: ctypes handle to libpq
+ hostaddr: If provided, use this specific address (e.g., "127.0.0.2")
+ port: If provided, use this port instead of finding a free one,
+ is currently only allowed if hostaddr is also provided
+ """
+
+ if hostaddr is None and port is not None:
+ raise NotImplementedError("port was provided without hostaddr")
+
+ self.name = name
+ self.datadir = datadir
+ self.sockdir = sockdir
+ self.libpq_handle = libpq_handle
+ self._remaining_timeout_fn: Optional[Callable[[], float]] = None
+ self._bindir = bindir
+ self._pg_ctl = bindir / "pg_ctl"
+ self.log = datadir / "postgresql.log"
+ self._log_start_pos = 0
+
+ # Determine whether to use Unix sockets
+ use_unix_sockets = platform.system() != "Windows" and hostaddr is None
+
+ # Use INITDB_TEMPLATE if available (much faster than running initdb)
+ initdb_template = os.environ.get("INITDB_TEMPLATE")
+ if initdb_template and os.path.isdir(initdb_template):
+ shutil.copytree(initdb_template, datadir)
+ else:
+ if platform.system() == "Windows":
+ auth_method = "trust"
+ else:
+ auth_method = "peer"
+ run(
+ bindir / "initdb",
+ "--no-sync",
+ "--auth",
+ auth_method,
+ "--pgdata",
+ self.datadir,
+ )
+
+ # Figure out a port to listen on. Attempt to reserve both IPv4 and IPv6
+ # addresses in one go.
+ #
+ # Note: socket.has_dualstack_ipv6/create_server are only in Python 3.8+.
+ if hostaddr is not None:
+ # Explicit address provided
+ addrs: list[str] = [hostaddr]
+ temp_sock = socket.socket()
+ if port is None:
+ temp_sock.bind((hostaddr, 0))
+ _, port = temp_sock.getsockname()
+
+ elif hasattr(socket, "has_dualstack_ipv6") and socket.has_dualstack_ipv6():
+ addr = ("::1", 0)
+ temp_sock = socket.create_server(
+ addr, family=socket.AF_INET6, dualstack_ipv6=True
+ )
+
+ hostaddr, port, _, _ = temp_sock.getsockname()
+ assert hostaddr is not None
+ addrs = [hostaddr, "127.0.0.1"]
+
+ else:
+ addr = ("127.0.0.1", 0)
+
+ temp_sock = socket.socket()
+ temp_sock.bind(addr)
+
+ hostaddr, port = temp_sock.getsockname()
+ assert hostaddr is not None
+ addrs = [hostaddr]
+
+ # Store the computed values
+ self.hostaddr = hostaddr
+ self.port = port
+ # Including the host to use for connections - either the socket
+ # directory or TCP address
+ if use_unix_sockets:
+ self.host = str(sockdir)
+ else:
+ self.host = hostaddr
+
+ with open(os.path.join(datadir, "postgresql.conf"), "a") as f:
+ print(file=f)
+ if use_unix_sockets:
+ print(
+ "unix_socket_directories = '{}'".format(sockdir.as_posix()),
+ file=f,
+ )
+ else:
+ # Disable Unix sockets when using TCP to avoid lock conflicts
+ print("unix_socket_directories = ''", file=f)
+ print("listen_addresses = '{}'".format(",".join(addrs)), file=f)
+ print("port =", port, file=f)
+ print("log_connections = all", file=f)
+ print("fsync = off", file=f)
+ print("datestyle = 'ISO'", file=f)
+ print("timezone = 'UTC'", file=f)
+
+ # Between closing of the socket, s, and server start, we're racing
+ # against anything that wants to open up ephemeral ports, so try not to
+ # put any new work here.
+
+ temp_sock.close()
+ self.pg_ctl("start")
+
+ # Read the PID file to get the postmaster PID
+ with open(os.path.join(datadir, "postmaster.pid")) as f:
+ self.pid = int(f.readline().strip())
+
+ # ExitStack for cleanup callbacks
+ self._cleanup_stack = contextlib.ExitStack()
+
+ def current_log_position(self):
+ """Get the current end position of the log file."""
+ if self.log.exists():
+ return self.log.stat().st_size
+ return 0
+
+ def reset_log_position(self):
+ """Mark current log position as start for log_content()."""
+ self._log_start_pos = self.current_log_position()
+
+ @contextlib.contextmanager
+ def start_new_test(self, remaining_timeout):
+ """
+ Prepare server for a new test.
+
+ Sets timeout, resets log position, and enters a cleanup subcontext.
+ """
+ self.set_timeout(remaining_timeout)
+ self.reset_log_position()
+ with self.subcontext():
+ yield self
+
+ def psql(self, *args):
+ """Run psql with the given arguments."""
+ self._run(os.path.join(self._bindir, "psql"), "-w", *args)
+
+ def sql(self, query):
+ """Execute a SQL query via libpq. Returns simplified results."""
+ with self.connect() as conn:
+ return conn.sql(query)
+
+ def pg_ctl(self, *args):
+ """Run pg_ctl with the given arguments."""
+ self._run(self._pg_ctl, "--pgdata", self.datadir, "--log", self.log, *args)
+
+ def _run(self, cmd, *args, addenv: Optional[dict] = None):
+ """Run a command with PG* environment variables set."""
+ subenv = dict(os.environ)
+ subenv.update(
+ {
+ "PGHOST": str(self.host),
+ "PGPORT": str(self.port),
+ "PGDATABASE": "postgres",
+ "PGDATA": str(self.datadir),
+ }
+ )
+ if addenv:
+ subenv.update(addenv)
+ run(cmd, *args, env=subenv)
+
+ def create_users(self, *userkeys: str):
+ """Create test users and register them for cleanup."""
+ usermap = {}
+ for u in userkeys:
+ name = u + "user"
+ usermap[u] = name
+ self.psql("-c", "CREATE USER " + name)
+ self._cleanup_stack.callback(self.psql, "-c", "DROP USER " + name)
+ return usermap
+
+ def create_dbs(self, *dbkeys: str):
+ """Create test databases and register them for cleanup."""
+ dbmap = {}
+ for d in dbkeys:
+ name = d + "db"
+ dbmap[d] = name
+ self.psql("-c", "CREATE DATABASE " + name)
+ self._cleanup_stack.callback(self.psql, "-c", "DROP DATABASE " + name)
+ return dbmap
+
+ @contextlib.contextmanager
+ def reloading(self):
+ """
+ Provides a context manager for making configuration changes.
+
+ If the context suite finishes successfully, the configuration will
+ be reloaded via pg_ctl. On teardown, the configuration changes will
+ be unwound, and the server will be signaled to reload again.
+
+ The context target contains the following attributes which can be
+ used to configure the server:
+ - .conf: modifies postgresql.conf
+ - .hba: modifies pg_hba.conf
+
+ For example:
+
+ with pg_server_session.reloading() as s:
+ s.conf.set(log_connections="on")
+ s.hba.prepend("local all all trust")
+ """
+ # Push a reload onto the stack before making any other
+ # unwindable changes. That way the order of operations will be
+ #
+ # # test
+ # - config change 1
+ # - config change 2
+ # - reload
+ # # teardown
+ # - undo config change 2
+ # - undo config change 1
+ # - reload
+ #
+ self._cleanup_stack.callback(self.pg_ctl, "reload")
+ yield self._backup_configuration()
+
+ # Now actually reload
+ self.pg_ctl("reload")
+
+ @contextlib.contextmanager
+ def restarting(self):
+ """Like .reloading(), but with a full server restart."""
+ self._cleanup_stack.callback(self.pg_ctl, "restart")
+ yield self._backup_configuration()
+ self.pg_ctl("restart")
+
+ def _backup_configuration(self):
+ # Wrap the existing HBA and configuration with FileBackups.
+ return Backup(
+ hba=self._cleanup_stack.enter_context(HBA(self.datadir)),
+ conf=self._cleanup_stack.enter_context(Config(self.datadir)),
+ )
+
+ @contextlib.contextmanager
+ def subcontext(self):
+ """
+ Create a new cleanup context for per-test isolation.
+
+ Temporarily replaces the cleanup stack so that any cleanup callbacks
+ registered within this context will be cleaned up when the context exits.
+ """
+ old_stack = self._cleanup_stack
+ self._cleanup_stack = contextlib.ExitStack()
+ try:
+ self._cleanup_stack.__enter__()
+ yield self
+ finally:
+ self._cleanup_stack.__exit__(None, None, None)
+ self._cleanup_stack = old_stack
+
+ def stop(self, mode="fast"):
+ """
+ Stop the PostgreSQL server instance.
+
+ Ignores failures if the server is already stopped.
+ """
+ try:
+ self.pg_ctl("stop", "--mode", mode)
+ except subprocess.CalledProcessError:
+ # Server may have already been stopped
+ pass
+
+ def log_content(self) -> str:
+ """Return log content from the current context's start position."""
+ with open(self.log) as f:
+ f.seek(self._log_start_pos)
+ return f.read()
+
+ @contextlib.contextmanager
+ def log_contains(self, pattern, times=None):
+ """
+ Context manager that checks if the log matches pattern during the block.
+
+ Args:
+ pattern: The regex pattern to search for.
+ times: If None, any number of matches is accepted.
+ If a number, exactly that many matches are required.
+ """
+ start_pos = self.current_log_position()
+ yield
+ with open(self.log) as f:
+ f.seek(start_pos)
+ content = f.read()
+ if times is None:
+ assert re.search(pattern, content), f"Pattern {pattern!r} not found in log"
+ else:
+ match_count = len(re.findall(pattern, content))
+ assert match_count == times, (
+ f"Expected {times} matches of {pattern!r}, found {match_count}"
+ )
+
+ def cleanup(self):
+ """Run all registered cleanup callbacks."""
+ self._cleanup_stack.close()
+
+ def set_timeout(self, remaining_timeout_fn: Callable[[], float]) -> None:
+ """
+ Set the timeout function for connections.
+ This is typically called by pg fixture for each test.
+ """
+ self._remaining_timeout_fn = remaining_timeout_fn
+
+ def connect(self, **opts) -> PGconn:
+ """
+ Creates a connection to this PostgreSQL server instance.
+
+ Args:
+ **opts: Additional connection options (can override defaults)
+
+ Returns:
+ PGconn: Connected database connection
+
+ Example:
+ conn = pg.connect()
+ conn = pg.connect(dbname='mydb')
+ """
+ if self._remaining_timeout_fn is None:
+ raise RuntimeError(
+ "Timeout function not set. Use set_timeout() or pg fixture."
+ )
+
+ defaults = {
+ "host": self.host,
+ "port": self.port,
+ "dbname": "postgres",
+ }
+ defaults.update(opts)
+
+ return libpq_connect(
+ self.libpq_handle,
+ self._cleanup_stack,
+ self._remaining_timeout_fn,
+ **defaults,
+ )
diff --git a/src/test/pytest/pypg/util.py b/src/test/pytest/pypg/util.py
new file mode 100644
index 00000000000..b2a1e627e4b
--- /dev/null
+++ b/src/test/pytest/pypg/util.py
@@ -0,0 +1,42 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import shlex
+import subprocess
+import sys
+
+
+def eprint(*args, **kwargs):
+ """eprint prints to stderr"""
+ print(*args, file=sys.stderr, **kwargs)
+
+
+def run(*command, check=True, shell=None, silent=False, **kwargs):
+ """run runs the given command and prints it to stderr"""
+
+ if shell is None:
+ shell = len(command) == 1 and isinstance(command[0], str)
+
+ if shell:
+ command = command[0]
+ else:
+ command = list(map(str, command))
+
+ if not silent:
+ if shell:
+ eprint(f"+ {command}")
+ else:
+ # We could normally use shlex.join here, but it's not available in
+ # Python 3.6 which we still like to support
+ unsafe_string_cmd = " ".join(map(shlex.quote, command))
+ eprint(f"+ {unsafe_string_cmd}")
+
+ if silent:
+ kwargs.setdefault("stdout", subprocess.DEVNULL)
+
+ return subprocess.run(command, check=check, shell=shell, **kwargs)
+
+
+def capture(command, *args, stdout=subprocess.PIPE, encoding="utf-8", **kwargs):
+ return run(
+ command, *args, stdout=stdout, encoding=encoding, **kwargs
+ ).stdout.removesuffix("\n")
diff --git a/src/test/pytest/pyt/conftest.py b/src/test/pytest/pyt/conftest.py
new file mode 100644
index 00000000000..dd73917c68c
--- /dev/null
+++ b/src/test/pytest/pyt/conftest.py
@@ -0,0 +1 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
diff --git a/src/test/pytest/pyt/test_errors.py b/src/test/pytest/pyt/test_errors.py
new file mode 100644
index 00000000000..771fe8f76e3
--- /dev/null
+++ b/src/test/pytest/pyt/test_errors.py
@@ -0,0 +1,34 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Tests for libpq error types and SQLSTATE-based exception mapping.
+"""
+
+import pytest
+from libpq import LibpqError
+
+
+def test_syntax_error(conn):
+ """Invalid SQL syntax raises LibpqError with correct SQLSTATE."""
+ with pytest.raises(LibpqError) as exc_info:
+ conn.sql("SELEC 1")
+
+ err = exc_info.value
+ assert err.sqlstate == "42601"
+ assert err.sqlstate_class == "42"
+ assert "syntax" in str(err).lower()
+
+
+def test_unique_violation(conn):
+ """Unique violation includes all error fields."""
+ conn.sql("CREATE TEMP TABLE test_uv (id int CONSTRAINT test_uv_pk PRIMARY KEY)")
+ conn.sql("INSERT INTO test_uv VALUES (1)")
+
+ with pytest.raises(LibpqError) as exc_info:
+ conn.sql("INSERT INTO test_uv VALUES (1)")
+
+ err = exc_info.value
+ assert err.sqlstate == "23505"
+ assert err.table_name == "test_uv"
+ assert err.constraint_name == "test_uv_pk"
+ assert err.detail == "Key (id)=(1) already exists."
diff --git a/src/test/pytest/pyt/test_libpq.py b/src/test/pytest/pyt/test_libpq.py
new file mode 100644
index 00000000000..1d0d9bc3b94
--- /dev/null
+++ b/src/test/pytest/pyt/test_libpq.py
@@ -0,0 +1,35 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+import contextlib
+import os
+import socket
+import struct
+import threading
+from typing import Callable
+
+import pytest
+
+from libpq import connstr, LibpqError
+
+
[email protected](
+ "opts, expected",
+ [
+ (dict(), ""),
+ (dict(port=5432), "port=5432"),
+ (dict(port=5432, dbname="postgres"), "port=5432 dbname=postgres"),
+ (dict(host=""), "host=''"),
+ (dict(host=" "), r"host=' '"),
+ (dict(keyword="'"), r"keyword=\'"),
+ (dict(keyword=" \\' "), r"keyword=' \\\' '"),
+ ],
+)
+def test_connstr(opts, expected):
+ """Tests the escape behavior for connstr()."""
+ assert connstr(opts) == expected
+
+
+def test_must_connect_errors(connect):
+ """Tests that connect() raises LibpqError."""
+ with pytest.raises(LibpqError, match="invalid connection option"):
+ connect(some_unknown_keyword="whatever")
diff --git a/src/test/pytest/pyt/test_multi_server.py b/src/test/pytest/pyt/test_multi_server.py
new file mode 100644
index 00000000000..8ee045b0cc8
--- /dev/null
+++ b/src/test/pytest/pyt/test_multi_server.py
@@ -0,0 +1,46 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Tests demonstrating multi-server functionality using create_pg fixture.
+
+These tests verify that the pytest infrastructure correctly handles
+multiple PostgreSQL server instances within a single test, and that
+module-scoped servers persist across tests.
+"""
+
+import pytest
+
+
+def test_multiple_servers_basic(create_pg):
+ """Test that we can create and connect to multiple servers."""
+ node1 = create_pg("primary")
+ node2 = create_pg("secondary")
+
+ conn1 = node1.connect()
+ conn2 = node2.connect()
+
+ # Each server should have its own data directory
+ datadir1 = conn1.sql("SHOW data_directory")
+ datadir2 = conn2.sql("SHOW data_directory")
+ assert datadir1 != datadir2
+
+ # Each server should be listening on a different port
+ assert node1.port != node2.port
+
+
[email protected](scope="module")
+def shared_server(create_pg_module):
+ """A server shared across all tests in this module."""
+ server = create_pg_module("shared")
+ server.sql("CREATE TABLE module_state (value int DEFAULT 0)")
+ return server
+
+
+def test_module_server_create_row(shared_server):
+ """First test: create a row in the shared server."""
+ shared_server.connect().sql("INSERT INTO module_state VALUES (42)")
+
+
+def test_module_server_see_row(shared_server):
+ """Second test: verify we see the row from the previous test."""
+ assert shared_server.connect().sql("SELECT value FROM module_state") == 42
diff --git a/src/test/pytest/pyt/test_query_helpers.py b/src/test/pytest/pyt/test_query_helpers.py
new file mode 100644
index 00000000000..abcd9084214
--- /dev/null
+++ b/src/test/pytest/pyt/test_query_helpers.py
@@ -0,0 +1,347 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Tests for query helper functions with type conversion and result simplification.
+"""
+
+import uuid
+
+import pytest
+
+
+def test_single_cell_int(conn):
+ """Single cell integer query returns just the value."""
+ result = conn.sql("SELECT 1")
+ assert result == 1
+ assert isinstance(result, int)
+
+
+def test_single_cell_string(conn):
+ """Single cell string query returns just the value."""
+ result = conn.sql("SELECT 'hello'")
+ assert result == "hello"
+ assert isinstance(result, str)
+
+
+def test_single_cell_bool(conn):
+ """Single cell boolean query returns just the value."""
+
+ result = conn.sql("SELECT true")
+ assert result is True
+ assert isinstance(result, bool)
+
+ result = conn.sql("SELECT false")
+ assert result is False
+
+
+def test_single_cell_float(conn):
+ """Single cell float query returns just the value."""
+
+ result = conn.sql("SELECT 3.14::float4")
+ assert isinstance(result, float)
+ assert abs(result - 3.14) < 0.01
+
+
+def test_single_cell_null(conn):
+ """Single cell NULL query returns None."""
+
+ result = conn.sql("SELECT NULL")
+ assert result is None
+
+
+def test_single_row_multiple_columns(conn):
+ """Single row with multiple columns returns a tuple."""
+
+ result = conn.sql("SELECT 1, 'hello', true")
+ assert result == (1, "hello", True)
+ assert isinstance(result, tuple)
+
+
+def test_single_column_multiple_rows(conn):
+ """Single column with multiple rows returns a list of values."""
+
+ result = conn.sql("SELECT * FROM generate_series(1, 3)")
+ assert result == [1, 2, 3]
+ assert isinstance(result, list)
+
+
+def test_multiple_rows_and_columns(conn):
+ """Multiple rows and columns returns list of tuples."""
+
+ result = conn.sql("SELECT * FROM (VALUES (1, 'a'), (2, 'b'), (3, 'c')) AS t")
+ assert result == [(1, "a"), (2, "b"), (3, "c")]
+ assert isinstance(result, list)
+ assert all(isinstance(row, tuple) for row in result)
+
+
+def test_empty_result(conn):
+ """Empty result set returns empty list."""
+
+ result = conn.sql("SELECT 1 WHERE false")
+ assert result == []
+
+
+def test_query_error_handling(conn):
+ """Query errors raise RuntimeError with actual error message."""
+
+ with pytest.raises(RuntimeError) as exc_info:
+ conn.sql("SELECT * FROM nonexistent_table")
+
+ error_msg = str(exc_info.value)
+ assert "nonexistent_table" in error_msg or "does not exist" in error_msg
+
+
+def test_division_by_zero_error(conn):
+ """Division by zero raises RuntimeError."""
+
+ with pytest.raises(RuntimeError) as exc_info:
+ conn.sql("SELECT 1/0")
+
+ error_msg = str(exc_info.value)
+ assert "division by zero" in error_msg.lower()
+
+
+def test_simple_exec_create_table(conn):
+ """sql for CREATE TABLE returns None."""
+
+ result = conn.sql("CREATE TEMP TABLE test_table (id int, name text)")
+ assert result is None
+
+ # Verify table was created
+ count = conn.sql("SELECT COUNT(*) FROM test_table")
+ assert count == 0
+
+
+def test_simple_exec_insert(conn):
+ """sql for INSERT returns None."""
+
+ conn.sql("CREATE TEMP TABLE test_table (id int, name text)")
+ result = conn.sql("INSERT INTO test_table VALUES (1, 'Alice'), (2, 'Bob')")
+ assert result is None
+
+ # Verify data was inserted
+ count = conn.sql("SELECT COUNT(*) FROM test_table")
+ assert count == 2
+
+
+def test_type_conversion_mixed(conn):
+ """Test mixed type conversion in a single row."""
+
+ result = conn.sql("SELECT 42::int4, 123::int8, 3.14::float8, 'text', true, NULL")
+ assert result == (42, 123, 3.14, "text", True, None)
+ assert isinstance(result[0], int)
+ assert isinstance(result[1], int)
+ assert isinstance(result[2], float)
+ assert isinstance(result[3], str)
+ assert isinstance(result[4], bool)
+ assert result[5] is None
+
+
+def test_multiple_queries_same_connection(conn):
+ """Test running multiple queries on the same connection."""
+
+ result1 = conn.sql("SELECT 1")
+ assert result1 == 1
+
+ result2 = conn.sql("SELECT 'hello', 'world'")
+ assert result2 == ("hello", "world")
+
+ result3 = conn.sql("SELECT * FROM generate_series(1, 5)")
+ assert result3 == [1, 2, 3, 4, 5]
+
+
+def test_date_type(conn):
+ """Test date type conversion."""
+ import datetime
+
+ result = conn.sql("SELECT '2025-10-20'::date")
+ assert result == datetime.date(2025, 10, 20)
+ assert isinstance(result, datetime.date)
+
+
+def test_timestamp_type(conn):
+ """Test timestamp type conversion."""
+ import datetime
+
+ result = conn.sql("SELECT '2025-10-20 15:30:45'::timestamp")
+ assert result == datetime.datetime(2025, 10, 20, 15, 30, 45)
+ assert isinstance(result, datetime.datetime)
+
+
+def test_time_type(conn):
+ """Test time type conversion."""
+ import datetime
+
+ result = conn.sql("SELECT '15:30:45'::time")
+ assert result == datetime.time(15, 30, 45)
+ assert isinstance(result, datetime.time)
+
+
+def test_numeric_type(conn):
+ """Test numeric/decimal type conversion."""
+ import decimal
+
+ result = conn.sql("SELECT 123.456::numeric")
+ assert result == decimal.Decimal("123.456")
+ assert isinstance(result, decimal.Decimal)
+
+
+def test_int_array(conn):
+ """Test integer array type conversion."""
+
+ result = conn.sql("SELECT ARRAY[1, 2, 3, 4, 5]")
+ assert result == [1, 2, 3, 4, 5]
+ assert isinstance(result, list)
+ assert all(isinstance(x, int) for x in result)
+
+
+def test_text_array(conn):
+ """Test text array type conversion."""
+
+ result = conn.sql("SELECT ARRAY['hello', 'world', 'test']")
+ assert result == ["hello", "world", "test"]
+ assert isinstance(result, list)
+ assert all(isinstance(x, str) for x in result)
+
+
+def test_bool_array(conn):
+ """Test boolean array type conversion."""
+
+ result = conn.sql("SELECT ARRAY[true, false, true]")
+ assert result == [True, False, True]
+ assert isinstance(result, list)
+ assert all(isinstance(x, bool) for x in result)
+
+
+def test_empty_array(conn):
+ """Test empty array type conversion."""
+
+ result = conn.sql("SELECT ARRAY[]::int[]")
+ assert result == []
+ assert isinstance(result, list)
+
+
+def test_json_type(conn):
+ """Test JSON type (parsed to dict)."""
+
+ result = conn.sql('SELECT \'{"key": "value"}\'::json')
+ assert isinstance(result, dict)
+ assert result == {"key": "value"}
+
+
+def test_jsonb_type(conn):
+ """Test JSONB type (parsed to dict)."""
+
+ result = conn.sql('SELECT \'{"name": "test", "count": 42}\'::jsonb')
+ assert isinstance(result, dict)
+ assert result == {"name": "test", "count": 42}
+
+
+def test_json_array(conn):
+ """Test JSON array type."""
+
+ result = conn.sql("SELECT '[1, 2, 3, 4, 5]'::json")
+ assert isinstance(result, list)
+ assert result == [1, 2, 3, 4, 5]
+
+
+def test_json_nested(conn):
+ """Test nested JSON object."""
+
+ result = conn.sql(
+ 'SELECT \'{"user": {"id": 1, "name": "Alice"}, "active": true}\'::json'
+ )
+ assert isinstance(result, dict)
+ assert result == {"user": {"id": 1, "name": "Alice"}, "active": True}
+
+
+def test_mixed_types_with_arrays(conn):
+ """Test mixed types including arrays in a single row."""
+
+ result = conn.sql("SELECT 42, 'text', ARRAY[1, 2, 3], true")
+ assert result == (42, "text", [1, 2, 3], True)
+ assert isinstance(result[0], int)
+ assert isinstance(result[1], str)
+ assert isinstance(result[2], list)
+ assert isinstance(result[3], bool)
+
+
+def test_uuid_type(conn):
+ """Test UUID type conversion."""
+ test_uuid = "550e8400-e29b-41d4-a716-446655440000"
+ result = conn.sql(f"SELECT '{test_uuid}'::uuid")
+ assert result == uuid.UUID(test_uuid)
+ assert isinstance(result, uuid.UUID)
+
+
+def test_uuid_generation(conn):
+ """Test generated UUID type conversion."""
+ result = conn.sql("SELECT uuidv4()")
+ assert isinstance(result, uuid.UUID)
+ # Check it's a valid UUID by ensuring it can be converted to string
+ assert len(str(result)) == 36 # UUID string format length
+
+
+def test_text_array_with_commas(conn):
+ """Test text array with elements containing commas."""
+
+ result = conn.sql("SELECT ARRAY['A,B', 'C', ' D ']")
+ assert result == ["A,B", "C", " D "]
+
+
+def test_text_array_with_quotes(conn):
+ """Test text array with elements containing quotes."""
+
+ result = conn.sql(r"SELECT ARRAY[E'a\"b', 'c']")
+ assert result == ['a"b', "c"]
+
+
+def test_text_array_with_backslash(conn):
+ """Test text array with elements containing backslashes."""
+
+ result = conn.sql(r"SELECT ARRAY[E'a\\b', 'c']")
+ assert result == ["a\\b", "c"]
+
+
+def test_json_array_type(conn):
+ """Test array of JSON values with embedded quotes and commas."""
+
+ result = conn.sql("""SELECT ARRAY['{"abc": 123, "xyz": 456}'::json]""")
+ assert result == [{"abc": 123, "xyz": 456}]
+
+
+def test_json_array_multiple(conn):
+ """Test array of multiple JSON objects."""
+
+ result = conn.sql(
+ """SELECT ARRAY['{"a": 1}'::json, '{"b": 2}'::json, '["x", "y"]'::json]"""
+ )
+ assert result == [{"a": 1}, {"b": 2}, ["x", "y"]]
+
+
+def test_2d_int_array(conn):
+ """Test 2D integer array."""
+
+ result = conn.sql("SELECT ARRAY[[1,2],[3,4]]")
+ assert result == [[1, 2], [3, 4]]
+
+
+def test_2d_text_array(conn):
+ """Test 2D integer array."""
+
+ result = conn.sql("SELECT ARRAY[['a','b'],['c','d,e']]")
+ assert result == [["a", "b"], ["c", "d,e"]]
+
+
+def test_3d_int_array(conn):
+ """Test 3D integer array."""
+
+ result = conn.sql("SELECT ARRAY[[[1,2],[3,4]],[[5,6],[7,8]]]")
+ assert result == [[[1, 2], [3, 4]], [[5, 6], [7, 8]]]
+
+
+def test_array_with_null(conn):
+ """Test array with NULL elements."""
+
+ result = conn.sql("SELECT ARRAY[1, NULL, 3]")
+ assert result == [1, None, 3]
--
2.53.0
[text/x-patch] v12-0003-POC-Convert-load-balance-tests-from-perl-to-pyth.patch (17.3K, 6-v12-0003-POC-Convert-load-balance-tests-from-perl-to-pyth.patch)
download | inline diff:
From b9db2925f41d32938cc81dfac3a56d8ce83bd48a Mon Sep 17 00:00:00 2001
From: Jelte Fennema-Nio <[email protected]>
Date: Fri, 26 Dec 2025 12:31:43 +0100
Subject: [PATCH v12 3/5] POC: Convert load balance tests from perl to python
This is a proof of concept to show how to use the pytest test
infrastructure. It converts two existing tests that could not share
code. And now they do. If we ever introduce another load balance method
(e.g. round robin). We can easily test it for both DNS and hostlist
based load balancing by adding a single new test function.
---
src/interfaces/libpq/Makefile | 1 +
src/interfaces/libpq/meson.build | 7 +-
src/interfaces/libpq/pyt/test_load_balance.py | 170 ++++++++++++++++++
.../libpq/t/003_load_balance_host_list.pl | 94 ----------
.../libpq/t/004_load_balance_dns.pl | 144 ---------------
5 files changed, 176 insertions(+), 240 deletions(-)
create mode 100644 src/interfaces/libpq/pyt/test_load_balance.py
delete mode 100644 src/interfaces/libpq/t/003_load_balance_host_list.pl
delete mode 100644 src/interfaces/libpq/t/004_load_balance_dns.pl
diff --git a/src/interfaces/libpq/Makefile b/src/interfaces/libpq/Makefile
index 0963995eed4..d088142f8c6 100644
--- a/src/interfaces/libpq/Makefile
+++ b/src/interfaces/libpq/Makefile
@@ -169,6 +169,7 @@ check installcheck: export PATH := $(CURDIR)/test:$(PATH)
check: test-build all
$(prove_check)
+ $(pytest_check)
installcheck: test-build all
$(prove_installcheck)
diff --git a/src/interfaces/libpq/meson.build b/src/interfaces/libpq/meson.build
index b0ae72167a1..62cde97d169 100644
--- a/src/interfaces/libpq/meson.build
+++ b/src/interfaces/libpq/meson.build
@@ -157,8 +157,6 @@ tests += {
'tests': [
't/001_uri.pl',
't/002_api.pl',
- 't/003_load_balance_host_list.pl',
- 't/004_load_balance_dns.pl',
't/005_negotiate_encryption.pl',
't/006_service.pl',
],
@@ -169,6 +167,11 @@ tests += {
},
'deps': libpq_test_deps,
},
+ 'pytest': {
+ 'tests': [
+ 'pyt/test_load_balance.py',
+ ],
+ },
}
subdir('po', if_found: libintl)
diff --git a/src/interfaces/libpq/pyt/test_load_balance.py b/src/interfaces/libpq/pyt/test_load_balance.py
new file mode 100644
index 00000000000..0af46d8f37d
--- /dev/null
+++ b/src/interfaces/libpq/pyt/test_load_balance.py
@@ -0,0 +1,170 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+"""
+Tests for load_balance_hosts connection parameter.
+
+These tests verify that libpq correctly handles load balancing across multiple
+PostgreSQL servers specified in the connection string.
+"""
+
+import platform
+import re
+
+import pytest
+
+from libpq import LibpqError
+import pypg
+
+
[email protected](scope="module")
+def load_balance_nodes_hostlist(create_pg_module):
+ """
+ Create 3 PostgreSQL nodes with different socket directories.
+
+ Each node has its own Unix socket directory for isolation.
+ Returns a tuple of (nodes, connect).
+ """
+ nodes = [create_pg_module() for _ in range(3)]
+
+ hostlist = ",".join(node.host for node in nodes)
+ portlist = ",".join(str(node.port) for node in nodes)
+
+ def connect(**kwargs):
+ return nodes[0].connect(host=hostlist, port=portlist, **kwargs)
+
+ return nodes, connect
+
+
[email protected](scope="module")
+def load_balance_nodes_dns(create_pg_module):
+ """
+ Create 3 PostgreSQL nodes on the same port but different IP addresses.
+
+ Uses 127.0.0.1, 127.0.0.2, 127.0.0.3 with a shared port, so that
+ connections to 'pg-loadbalancetest' can be load balanced via DNS.
+
+ Since setting up a DNS server is more effort than we consider reasonable to
+ run this test, this situation is instead imitated by using a hosts file
+ where a single hostname maps to multiple different IP addresses. This test
+ requires the administrator to add the following lines to the hosts file (if
+ we detect that this hasn't happened we skip the test):
+
+ 127.0.0.1 pg-loadbalancetest
+ 127.0.0.2 pg-loadbalancetest
+ 127.0.0.3 pg-loadbalancetest
+
+ Windows or Linux are required to run this test because these OSes allow
+ binding to 127.0.0.2 and 127.0.0.3 addresses by default, but other OSes
+ don't. We need to bind to different IP addresses, so that we can use these
+ different IP addresses in the hosts file.
+
+ The hosts file needs to be prepared before running this test. We don't do
+ it on the fly, because it requires root permissions to change the hosts
+ file. In CI we set up the previously mentioned rules in the hosts file, so
+ that this load balancing method is tested.
+
+ Requires PG_TEST_EXTRA=load_balance because it requires this manual hosts
+ file configuration and also uses TCP with trust auth, which is potentially
+ unsafe on multiuser systems.
+ """
+ pypg.skip_unless_test_extras("load_balance")
+
+ if platform.system() not in ("Linux", "Windows"):
+ pytest.skip("DNS load balance test only supported on Linux and Windows")
+
+ if platform.system() == "Windows":
+ hosts_path = r"c:\Windows\System32\Drivers\etc\hosts"
+ else:
+ hosts_path = "/etc/hosts"
+
+ try:
+ with open(hosts_path) as f:
+ hosts_content = f.read()
+ except (OSError, IOError):
+ pytest.skip(f"Could not read hosts file: {hosts_path}")
+
+ count = len(re.findall(r"127\.0\.0\.[1-3]\s+pg-loadbalancetest", hosts_content))
+ if count != 3:
+ pytest.skip("hosts file not prepared for DNS load balance test")
+
+ first_node = create_pg_module(hostaddr="127.0.0.1")
+ nodes = [
+ first_node,
+ create_pg_module(hostaddr="127.0.0.2", port=first_node.port),
+ create_pg_module(hostaddr="127.0.0.3", port=first_node.port),
+ ]
+
+ # Allow trust authentication for TCP connections from loopback
+ for node in nodes:
+ hba_path = node.datadir / "pg_hba.conf"
+ with open(hba_path, "r") as f:
+ original_content = f.read()
+ with open(hba_path, "w") as f:
+ f.write("host all all 127.0.0.0/8 trust\n")
+ f.write(original_content)
+ node.pg_ctl("reload")
+
+ def connect(**kwargs):
+ return nodes[0].connect(host="pg-loadbalancetest", **kwargs)
+
+ return nodes, connect
+
+
[email protected](scope="module", params=["hostlist", "dns"])
+def load_balance_nodes(request):
+ """
+ Parametrized fixture providing both load balancing test environments.
+ """
+ return request.getfixturevalue(f"load_balance_nodes_{request.param}")
+
+
+def test_load_balance_hosts_invalid_value(load_balance_nodes):
+ """load_balance_hosts doesn't accept unknown values."""
+ _, connect = load_balance_nodes
+
+ with pytest.raises(
+ LibpqError, match='invalid load_balance_hosts value: "doesnotexist"'
+ ):
+ connect(load_balance_hosts="doesnotexist")
+
+
+def test_load_balance_hosts_disable(load_balance_nodes):
+ """load_balance_hosts=disable always connects to the first node."""
+ nodes, connect = load_balance_nodes
+
+ with nodes[0].log_contains("connection received"):
+ connect(load_balance_hosts="disable")
+
+
+def test_load_balance_hosts_random_distribution(load_balance_nodes):
+ """load_balance_hosts=random distributes connections across all nodes."""
+ nodes, connect = load_balance_nodes
+
+ for _ in range(50):
+ connect(load_balance_hosts="random")
+
+ occurrences = [
+ len(re.findall("connection received", node.log_content())) for node in nodes
+ ]
+
+ # Statistically, each node should receive at least one connection.
+ # The probability of any node receiving 0 connections is (2/3)^50 ≈ 1.57e-9
+ assert occurrences[0] > 0, "node1 should receive at least one connection"
+ assert occurrences[1] > 0, "node2 should receive at least one connection"
+ assert occurrences[2] > 0, "node3 should receive at least one connection"
+ assert sum(occurrences) == 50, "total connections should be 50"
+
+
+def test_load_balance_hosts_failover(load_balance_nodes):
+ """load_balance_hosts continues trying hosts until it finds a working one."""
+ nodes, connect = load_balance_nodes
+
+ nodes[0].stop()
+ nodes[1].stop()
+
+ with nodes[2].log_contains("connection received"):
+ connect(load_balance_hosts="disable")
+
+ with nodes[2].log_contains("connection received", times=5):
+ for _ in range(5):
+ connect(load_balance_hosts="random")
diff --git a/src/interfaces/libpq/t/003_load_balance_host_list.pl b/src/interfaces/libpq/t/003_load_balance_host_list.pl
deleted file mode 100644
index 1f970ff994b..00000000000
--- a/src/interfaces/libpq/t/003_load_balance_host_list.pl
+++ /dev/null
@@ -1,94 +0,0 @@
-# Copyright (c) 2023-2026, PostgreSQL Global Development Group
-use strict;
-use warnings FATAL => 'all';
-use Config;
-use PostgreSQL::Test::Utils;
-use PostgreSQL::Test::Cluster;
-use Test::More;
-
-# This tests load balancing across the list of different hosts in the host
-# parameter of the connection string.
-
-# Cluster setup which is shared for testing both load balancing methods
-my $node1 = PostgreSQL::Test::Cluster->new('node1');
-my $node2 = PostgreSQL::Test::Cluster->new('node2', own_host => 1);
-my $node3 = PostgreSQL::Test::Cluster->new('node3', own_host => 1);
-
-# Create a data directory with initdb
-$node1->init();
-$node2->init();
-$node3->init();
-
-# Start the PostgreSQL server
-$node1->start();
-$node2->start();
-$node3->start();
-
-# Start the tests for load balancing method 1
-my $hostlist = $node1->host . ',' . $node2->host . ',' . $node3->host;
-my $portlist = $node1->port . ',' . $node2->port . ',' . $node3->port;
-
-$node1->connect_fails(
- "host=$hostlist port=$portlist load_balance_hosts=doesnotexist",
- "load_balance_hosts doesn't accept unknown values",
- expected_stderr => qr/invalid load_balance_hosts value: "doesnotexist"/);
-
-# load_balance_hosts=disable should always choose the first one.
-$node1->connect_ok(
- "host=$hostlist port=$portlist load_balance_hosts=disable",
- "load_balance_hosts=disable connects to the first node",
- sql => "SELECT 'connect1'",
- log_like => [qr/statement: SELECT 'connect1'/]);
-
-# Statistically the following loop with load_balance_hosts=random will almost
-# certainly connect at least once to each of the nodes. The chance of that not
-# happening is so small that it's negligible: (2/3)^50 = 1.56832855e-9
-foreach my $i (1 .. 50)
-{
- $node1->connect_ok(
- "host=$hostlist port=$portlist load_balance_hosts=random",
- "repeated connections with random load balancing",
- sql => "SELECT 'connect2'");
-}
-
-my $node1_occurrences = () =
- $node1->log_content() =~ /statement: SELECT 'connect2'/g;
-my $node2_occurrences = () =
- $node2->log_content() =~ /statement: SELECT 'connect2'/g;
-my $node3_occurrences = () =
- $node3->log_content() =~ /statement: SELECT 'connect2'/g;
-
-my $total_occurrences =
- $node1_occurrences + $node2_occurrences + $node3_occurrences;
-
-cmp_ok($node1_occurrences, '>', 1,
- "received at least one connection on node1");
-cmp_ok($node2_occurrences, '>', 1,
- "received at least one connection on node2");
-cmp_ok($node3_occurrences, '>', 1,
- "received at least one connection on node3");
-is($total_occurrences, 50, "received 50 connections across all nodes");
-
-$node1->stop();
-$node2->stop();
-
-# load_balance_hosts=disable should continue trying hosts until it finds a
-# working one.
-$node3->connect_ok(
- "host=$hostlist port=$portlist load_balance_hosts=disable",
- "load_balance_hosts=disable continues until it connects to the a working node",
- sql => "SELECT 'connect3'",
- log_like => [qr/statement: SELECT 'connect3'/]);
-
-# Also with load_balance_hosts=random we continue to the next nodes if previous
-# ones are down. Connect a few times to make sure it's not just lucky.
-foreach my $i (1 .. 5)
-{
- $node3->connect_ok(
- "host=$hostlist port=$portlist load_balance_hosts=random",
- "load_balance_hosts=random continues until it connects to the a working node",
- sql => "SELECT 'connect4'",
- log_like => [qr/statement: SELECT 'connect4'/]);
-}
-
-done_testing();
diff --git a/src/interfaces/libpq/t/004_load_balance_dns.pl b/src/interfaces/libpq/t/004_load_balance_dns.pl
deleted file mode 100644
index e1ff9a06024..00000000000
--- a/src/interfaces/libpq/t/004_load_balance_dns.pl
+++ /dev/null
@@ -1,144 +0,0 @@
-# Copyright (c) 2023-2026, PostgreSQL Global Development Group
-use strict;
-use warnings FATAL => 'all';
-use Config;
-use PostgreSQL::Test::Utils;
-use PostgreSQL::Test::Cluster;
-use Test::More;
-
-if (!$ENV{PG_TEST_EXTRA} || $ENV{PG_TEST_EXTRA} !~ /\bload_balance\b/)
-{
- plan skip_all =>
- 'Potentially unsafe test load_balance not enabled in PG_TEST_EXTRA';
-}
-
-# This tests loadbalancing based on a DNS entry that contains multiple records
-# for different IPs. Since setting up a DNS server is more effort than we
-# consider reasonable to run this test, this situation is instead imitated by
-# using a hosts file where a single hostname maps to multiple different IP
-# addresses. This test requires the administrator to add the following lines to
-# the hosts file (if we detect that this hasn't happened we skip the test):
-#
-# 127.0.0.1 pg-loadbalancetest
-# 127.0.0.2 pg-loadbalancetest
-# 127.0.0.3 pg-loadbalancetest
-#
-# Windows or Linux are required to run this test because these OSes allow
-# binding to 127.0.0.2 and 127.0.0.3 addresses by default, but other OSes
-# don't. We need to bind to different IP addresses, so that we can use these
-# different IP addresses in the hosts file.
-#
-# The hosts file needs to be prepared before running this test. We don't do it
-# on the fly, because it requires root permissions to change the hosts file. In
-# CI we set up the previously mentioned rules in the hosts file, so that this
-# load balancing method is tested.
-
-# Cluster setup which is shared for testing both load balancing methods
-my $can_bind_to_127_0_0_2 =
- $Config{osname} eq 'linux' || $PostgreSQL::Test::Utils::windows_os;
-
-# Checks for the requirements for testing load balancing method 2
-if (!$can_bind_to_127_0_0_2)
-{
- plan skip_all => 'load_balance test only supported on Linux and Windows';
-}
-
-my $hosts_path;
-if ($windows_os)
-{
- $hosts_path = 'c:\Windows\System32\Drivers\etc\hosts';
-}
-else
-{
- $hosts_path = '/etc/hosts';
-}
-
-my $hosts_content = PostgreSQL::Test::Utils::slurp_file($hosts_path);
-
-my $hosts_count = () =
- $hosts_content =~ /127\.0\.0\.[1-3] pg-loadbalancetest/g;
-if ($hosts_count != 3)
-{
- # Host file is not prepared for this test
- plan skip_all => "hosts file was not prepared for DNS load balance test";
-}
-
-$PostgreSQL::Test::Cluster::use_tcp = 1;
-$PostgreSQL::Test::Cluster::test_pghost = '127.0.0.1';
-my $port = PostgreSQL::Test::Cluster::get_free_port();
-my $node1 = PostgreSQL::Test::Cluster->new('node1', port => $port);
-my $node2 =
- PostgreSQL::Test::Cluster->new('node2', port => $port, own_host => 1);
-my $node3 =
- PostgreSQL::Test::Cluster->new('node3', port => $port, own_host => 1);
-
-# Create a data directory with initdb
-$node1->init();
-$node2->init();
-$node3->init();
-
-# Start the PostgreSQL server
-$node1->start();
-$node2->start();
-$node3->start();
-
-# load_balance_hosts=disable should always choose the first one.
-$node1->connect_ok(
- "host=pg-loadbalancetest port=$port load_balance_hosts=disable",
- "load_balance_hosts=disable connects to the first node",
- sql => "SELECT 'connect1'",
- log_like => [qr/statement: SELECT 'connect1'/]);
-
-
-# Statistically the following loop with load_balance_hosts=random will almost
-# certainly connect at least once to each of the nodes. The chance of that not
-# happening is so small that it's negligible: (2/3)^50 = 1.56832855e-9
-foreach my $i (1 .. 50)
-{
- $node1->connect_ok(
- "host=pg-loadbalancetest port=$port load_balance_hosts=random",
- "repeated connections with random load balancing",
- sql => "SELECT 'connect2'");
-}
-
-my $node1_occurrences = () =
- $node1->log_content() =~ /statement: SELECT 'connect2'/g;
-my $node2_occurrences = () =
- $node2->log_content() =~ /statement: SELECT 'connect2'/g;
-my $node3_occurrences = () =
- $node3->log_content() =~ /statement: SELECT 'connect2'/g;
-
-my $total_occurrences =
- $node1_occurrences + $node2_occurrences + $node3_occurrences;
-
-cmp_ok($node1_occurrences, '>', 1,
- "received at least one connection on node1");
-cmp_ok($node2_occurrences, '>', 1,
- "received at least one connection on node2");
-cmp_ok($node3_occurrences, '>', 1,
- "received at least one connection on node3");
-is($total_occurrences, 50, "received 50 connections across all nodes");
-
-$node1->stop();
-$node2->stop();
-
-# load_balance_hosts=disable should continue trying hosts until it finds a
-# working one.
-$node3->connect_ok(
- "host=pg-loadbalancetest port=$port load_balance_hosts=disable",
- "load_balance_hosts=disable continues until it connects to a working node",
- sql => "SELECT 'connect3'",
- log_like => [qr/statement: SELECT 'connect3'/]);
-
-# Also with load_balance_hosts=random we continue to the next nodes if previous
-# ones are down. Connect a few times to make sure it's not just lucky.
-foreach my $i (1 .. 5)
-{
- $node3->connect_ok(
- "host=pg-loadbalancetest port=$port load_balance_hosts=random",
- "load_balance_hosts=random continues until it connects to a working node",
- sql => "SELECT 'connect4'",
- log_like => [qr/statement: SELECT 'connect4'/]);
-}
-
-done_testing();
--
2.53.0
view thread (23+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: RFC: adding pytest as a supported test framework
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox