#python.src.stricttest.pgcluster

#python.src.stricttest.pgcluster

An ephemeral PostgreSQL cluster for test suites.

This is a cluster launcher, one layer below the per-test database managers that consumers already have: it boots a real postmaster on a private tmpfs directory with fsync off, hands back a base libpq URL, and shuts the cluster down again. Creating and dropping a database per test is then an ordinary CREATE DATABASE against that URL.

The model is one shared cluster per session, many ephemeral databases inside it. Booting a cluster costs roughly a second; creating a database inside a running one costs milliseconds. A per-test cluster would be neither.

A consumer writes its own fixtures -- the plugin deliberately ships none, so that a suite with no database pays nothing::

import pytest from stricttest.pgcluster import ephemeral_cluster

@pytest.fixture(scope="session") def pg(): with ephemeral_cluster(dsn_env="MYAPP_DATABASE_URL") as cluster: yield cluster

@pytest.fixture def db_url(pg): with pg.database(export=True) as url: yield url

There is no default name for dsn_env: the variable an application reads its connection string from is the application's decision, and guessing it would either do nothing or silently point production configuration at a test cluster.

Nothing here is imported by the plugin itself. The module depends only on the standard library and the PostgreSQL binaries it launches -- initdb, pg_ctl and psql -- so the plugin keeps its "pytest and the standard library, nothing else" rule.

Interaction with the socket guard. The cluster listens on a unix socket only. What that means for the socket guard depends entirely on how the driver is implemented, and the answer is not the one an allowlist key suggests.

asyncpg speaks the wire protocol from Python and opens its connection through socket, so the audit hook fires and the guard refuses the connect unless it is allowlisted. A suite that reaches the cluster through asyncpg must allowlist the socket directory's parent as a prefix::

stricttest_unix_socket_allowlist = ["/dev/shm/"]

psycopg -- and anything else built on libpq -- is a C extension. The connect happens inside libpq, never through Python's socket module, so no audit event is ever raised. The guard does not see the connection, cannot refuse it, and cannot be made to allow it: there is no event to allow. Adding the socket directory to stricttest_unix_socket_allowlist changes nothing for a libpq consumer, in either direction, and no stance this plugin offers protects one. The same is true of psql, which is a subprocess.

The protection that does work for every driver is this module. Point the application's DSN at the ephemeral cluster and a connection the guard never saw still lands in a throwaway database on a private socket rather than on a real server. That is structural rather than a policy, which is why it holds for the drivers the guard is blind to.

#PostgresUnavailable

No usable PostgreSQL installation was found on this machine.

Raised by :func:find_binaries and by :meth:EphemeralCluster.start. It carries a precise reason so a test can skip with it verbatim::

try: binaries = find_binaries() except PostgresUnavailable as exc: pytest.skip(str(exc))

#SocketPathTooLong

A socket directory would produce a path past the kernel's sun_path limit.

#PostgresClusterError

A PostgreSQL program this module ran failed.

The message carries the command, its output, and the server log when one exists -- a bind() failure or a refused start says nothing useful on its own.

#Binaries

Resolved paths to the PostgreSQL programs this module runs.

#bindir

def bindir(self) -> Path

#_search_dirs

def _search_dirs(extra_dirs: Sequence[str | Path]) -> list[Path]

Expand the search-dir globs into existing directories, newest first.

#find_binaries

def find_binaries(extra_dirs: Sequence[str | Path]=()) -> Binaries

Locate initdb, pg_ctl and psql.

PATH wins; otherwise the layouts in :data:BINARY_SEARCH_DIRS are searched in order. extra_dirs is searched before either. Raises :class:PostgresUnavailable naming exactly what was missing and where it was looked for.

#socket_path_for

def socket_path_for(directory: str | Path, port: int) -> str

The socket file PostgreSQL will create for port in directory.

#check_socket_dir

def check_socket_dir(directory: str | Path, port: int) -> str

Validate a socket directory, returning the socket path it would produce.

Raises :class:SocketPathTooLong when the resulting socket path would not fit in sockaddr_un.sun_path, and :class:ValueError when the path contains whitespace (the directory is passed to the postmaster inside a space-separated options string, where a space would silently split it).

#_pick_parent

def _pick_parent(explicit: str | Path | None, candidates: Sequence[str], *, port: int, check_socket: bool, what: str) -> Path

Choose a parent directory, explicitly or from the candidate list.

An explicit choice is never second-guessed: if it is unusable, that is an error, not a reason to quietly use something else. Without one, the candidates are tried in their fixed order and the first usable one wins.

#EphemeralCluster

A throwaway PostgreSQL cluster on a private tmpfs directory.

Use :func:ephemeral_cluster rather than constructing and starting this directly unless the start and stop must be separated.

#running

def running(self) -> bool

#socket_dir

def socket_dir(self) -> Path

The directory holding the cluster's unix socket.

#data_dir

def data_dir(self) -> Path

#socket_path

def socket_path(self) -> str

The full path of the cluster's unix socket file.

#base_url

def base_url(self) -> str

The libpq URL of the maintenance database.

This is what gets exported under dsn_env, and what a per-test database manager connects to in order to CREATE DATABASE.

#url_for

def url_for(self, dbname: str) -> str

The libpq URL of dbname in this cluster.

#start

def start(self) -> EphemeralCluster

initdb into tmpfs, start the postmaster, export the DSN.

#stop

def stop(self) -> None

Stop the postmaster and delete everything it wrote (idempotent).

#generate_name

def generate_name() -> str

A fresh, unique, always-acceptable database name.

#create_database

def create_database(self, name: str | None=None) -> str

Create a database in this cluster and return its libpq URL.

Without a name, a fresh unique one is generated -- the ordinary per-test case.

#drop_database

def drop_database(self, name: str) -> None

Drop a database, disconnecting anything still attached to it.

#database

def database(self, name: str | None=None, *, export: bool=False) -> Iterator[str]

Create a database for the block's duration and drop it afterwards.

With export=True the database's URL replaces the cluster's base URL under dsn_env for the duration, so an application that reads its connection string from the environment lands in the per-test database.

#sql

def sql(self, statement: str, *, dbname: str | None=None) -> str

Run one SQL statement through psql and return its output.

A subprocess rather than a driver on purpose: the plugin depends on pytest and the standard library only, and a database driver would be a third dependency every consumer inherits.

#_env

def _env(self) -> dict[str, str]

A clean environment for the PostgreSQL programs.

Every PG* variable is dropped: an ambient PGHOST, PGDATABASE, PGSERVICE or PGPASSFILE would silently redirect these commands at the developer's real cluster. The locale is pinned so error messages are the ones this module's callers expect to match on.

#ephemeral_cluster

def ephemeral_cluster(dsn_env: str, **kwargs) -> Iterator[EphemeralCluster]

Start a throwaway cluster for the block's duration.

dsn_env names the environment variable the base URL is exported under; every other argument is :class:EphemeralCluster's. The cluster is stopped and its directories removed on the way out, including when the block raises.