PostgreSQL testing utility. Python 3.7.17+ is supported.

Utility for orchestrating temporary PostgreSQL clusters in Python tests. Supports Python 3.7.17 and newer.

To install`testgres`, run:

Install`testgres` from PyPI:

pip install testgres

We encourage you to use`virtualenv` for your testing environment.

Use a dedicated virtual environment for isolated test dependencies.

There are several ways to specify a custom postgres installation:

Specify a custom PostgreSQL installation in one of the following ways:

* export`PG_CONFIG` environment variablepointingto the`pg_config` executable;

* export`PG_BIN` environment variablepointingto the directory withexecutable files.

- Set the`PG_CONFIG` environment variableto pointto the`pg_config` executable.

- Set the`PG_BIN` environment variableto pointto the directory withPostgreSQL binaries.

Example:

export PG_BIN=$HOME/pg_10/bin

export PG_BIN=$HOME/pg_16/bin

python my_tests.py

Here is an example of what you can do with`testgres`:

Create a temporary node, run queries, and let`testgres` clean up automatically:

with testgres.get_new_node()as node:

node.init()

node.start()

print(node.execute('select 1'))

There are four API methods for running queries:

`testgres` provides four helpers for executing queries against the node:

| Command| Description|

|----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|

|`node.psql(query, ...)`| Runs query via`psql` command and returns tuple`(error code, stdout, stderr)`.|

|`node.safe_psql(query, ...)`| Same as`psql()` except that it returns only`stdout`. If an error occurs during the execution, an exception will be thrown.|

|`node.execute(query, ...)`| Connects to PostgreSQL using`psycopg2` or`pg8000` (depends on which one is installed in your system) and returns two-dimensional array with data.|

|`node.connect(dbname, ...)`| Returns connection wrapper (`NodeConnection`) capable of running several queries within a single transaction.|

|---------|-------------|

|`node.psql(query, ...)`| Runs the query via`psql` and returns a tuple`(returncode, stdout, stderr)`.|

|`node.safe_psql(query, ...)`| Same as`psql()` but returns only`stdout` and raises if the command fails.|

|`node.execute(query, ...)`| Connects via`psycopg2` or`pg8000` (whichever is available) and returns a list of tuples.|

|`node.connect(dbname, ...)`| Returns a`NodeConnection` wrapper for executing multiple statements within a transaction.|

Example of transactional usage:

The last one is the most powerful: you can use`begin(isolation_level)`,`commit()` and`rollback()`:

with node.connect()as con:

con.begin('serializable')

print(con.execute('select%s',1))

con.rollback()

By default,`cleanup()` removes all temporary files (DB files, logs etc) that were created by testgres' API methods.

If you'd like to keep logs, execute`configure_testgres(node_cleanup_full=False)` before running any tests.

By default`cleanup()` removes all temporary files (data directories, logs, and so on) created by the API. Call`configure_testgres(node_cleanup_full=False)` before starting nodes if you want to keep logs for inspection.

`testgres` supports[python logging](https://docs.python.org/3.6/library/logging.html),

which means that you can aggregate logs from several nodes into one file:

`testgres` integrates with the standard[Python logging](https://docs.python.org/3/library/logging.html) module, so you can aggregate logs from multiple nodes:

import logging

logging.basicConfig(filename='/tmp/testgres.log')

testgres.configure_testgres(use_python_logging=True)

node1= testgres.get_new_node().init().start()

node2= testgres.get_new_node().init().start()

node1.execute('select 1')

node2.execute('select 2')

testgres.configure_testgres(use_python_logging=False)

Look at`tests/test_simple.py` file for a complete example of the logging

configuration.

See`tests/test_simple.py` for a complete logging example.

It's quite easy to create a backupandstart a new replica:

Creating backupsandspawning replicas is straightforward:

with testgres.get_new_node('master')as master:

master.init().start()

with master.backup()as backup:

replica= backup.spawn_replica('replica').start()

replica.catchup()

print(replica.execute('postgres','select 1'))

`testgres` is also capable of running benchmarks using`pgbench`:

Use`pgbench` through`testgres` to run quick benchmarks:

with testgres.get_new_node('master')as master:

master.init().start()

res= master.pgbench_init(scale=2).pgbench_run(time=10)

print(res)

result= master.pgbench_init(scale=2).pgbench_run(time=10)

print(result)

It's often useful to extend default configuration provided by`testgres`.

`testgres` has`default_conf()` function that helps control some basic

options. The`append_conf()` function can be used to add custom

lines to configuration lines:

`testgres` ships with sensible defaults. Adjust them as needed with`default_conf()` and`append_conf()`:

with testgres.get_new_node().init()as master:

master.default_conf(fsync=True,

allow_streaming=True)

master.append_conf('postgresql.conf', ext_conf)

master.default_conf(fsync=True,allow_streaming=True)

master.append_conf('postgresql.conf', extra_conf)

Note that`default_conf()` is called by`init()` function; both of them overwrite

the configuration file, which means that they should be called before`append_conf()`.

`default_conf()` is called by`init()` and rewrites the configuration file. Apply`append_conf()` afterwards to keep custom lines.

Testgres supports the creation of PostgreSQL nodes on a remote host. This is useful when you want to run distributed tests involving multiple nodes spread across different machines.

To use this feature, you need to use the RemoteOperations class. This feature is only supported with Linux.

Here is an example of how you might set this up:

You can provision nodes on a remote host (Linux only) by wiring`RemoteOperations` into the configuration:

from testgresimport ConnectionParams, RemoteOperations, TestgresConfig, get_remote_node

conn_params= ConnectionParams(

)

os_ops= RemoteOperations(conn_params)

TestgresConfig.set_os_ops(os_ops=os_ops)

deftest_basic_query(self):

deftest_basic_query():

with get_remote_node(conn_params=conn_params)as node:

node.init().start()

res= node.execute('SELECT 1')

self.assertEqual(res, [(1,)])

assert node.execute('SELECT 1')== [(1,)]

Use fixtures to create and clean up nodes automatically when testing with`pytest`:

import pytest

import testgres

defpg_node():

node= testgres.get_new_node().init().start()

try:

yield node

finally:

node.stop()

node.cleanup()

deftest_simple(pg_node):

assert pg_node.execute('select 1')[0][0]==1

This pattern keeps tests concise and ensures that every node is stopped and removed even if the test fails.

- Run tests in parallel with`pytest -n auto` (requires`pytest-xdist`). Ensure each node uses a distinct port by setting`PGPORT` in the fixture or by passing the`port` argument to`get_new_node()`.

- Always call`node.cleanup()` after each test, or rely on context managers/fixtures that do it for you, to avoid leftover data directories.

- Prefer`node.safe_psql()` for lightweight assertions that should fail fast; use`node.execute()` when you need structured Python results.

[Ildar Musin](https://github.com/zilder)

[Dmitry Ivanov](https://github.com/funbringer)

[Ildus Kurbangaliev](https://github.com/ildus)

[Yury Zhuravlev](https://github.com/stalkerg)

[Yury Zhuravlev](https://github.com/stalkerg)