Repository files navigation

Documentation

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

Installtestgres from PyPI:

pip install testgres

Use a dedicated virtual environment for isolated test dependencies.

Note: by defaulttestgres invokesinitdb,pg_ctl, andpsql binaries found inPATH.

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

• Set thePG_CONFIG environment variable to point to thepg_config executable.
• Set thePG_BIN environment variable to point to the directory with PostgreSQL binaries.

Example:

export PG_BIN=$HOME/pg_16/binpython my_tests.py

Create a temporary node, run queries, and lettestgres clean up automatically:

# create a node with a random name, port, and data directorywithtestgres.get_new_node()asnode:# run initdbnode.init()# start PostgreSQLnode.start()# execute a query in the default databaseprint(node.execute('select 1'))# the node is stopped and its files are removed automatically

testgres provides four helpers for executing queries against the node:

Example of transactional usage:

withnode.connect()ascon:con.begin('serializable')print(con.execute('select %s',1))con.rollback()

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

Note: context managers (thewith statement) callstop() andcleanup() automatically.

testgres integrates with the standardPython logging module, so you can aggregate logs from multiple nodes:

importlogging# write everything to /tmp/testgres.loglogging.basicConfig(filename='/tmp/testgres.log')# enable logging and create two nodestestgres.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')# disable loggingtestgres.configure_testgres(use_python_logging=False)

Seetests/test_simple.py for a complete logging example.

Creating backups and spawning replicas is straightforward:

withtestgres.get_new_node('master')asmaster:master.init().start()withmaster.backup()asbackup:replica=backup.spawn_replica('replica').start()replica.catchup()print(replica.execute('postgres','select 1'))

Usepgbench throughtestgres to run quick benchmarks:

withtestgres.get_new_node('master')asmaster:master.init().start()result=master.pgbench_init(scale=2).pgbench_run(time=10)print(result)

testgres ships with sensible defaults. Adjust them as needed withdefault_conf() andappend_conf():

extra_conf="shared_preload_libraries = 'postgres_fdw'"withtestgres.get_new_node().init()asmaster:master.default_conf(fsync=True,allow_streaming=True)master.append_conf('postgresql.conf',extra_conf)

default_conf() is called byinit() and rewrites the configuration file. Applyappend_conf() afterwards to keep custom lines.

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

fromtestgresimportConnectionParams,RemoteOperations,TestgresConfig,get_remote_nodeconn_params=ConnectionParams(host='example.com',username='postgres',ssh_key='/path/to/ssh/key')os_ops=RemoteOperations(conn_params)TestgresConfig.set_os_ops(os_ops=os_ops)deftest_basic_query():withget_remote_node(conn_params=conn_params)asnode:node.init().start()assertnode.execute('SELECT 1')== [(1,)]

Use fixtures to create and clean up nodes automatically when testing withpytest:

importpytestimporttestgres@pytest.fixturedefpg_node():node=testgres.get_new_node().init().start()try:yieldnodefinally:node.stop()node.cleanup()deftest_simple(pg_node):assertpg_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 withpytest -n auto (requirespytest-xdist). Ensure each node uses a distinct port by settingPGPORT in the fixture or by passing theport argument toget_new_node().
• Always callnode.cleanup() after each test, or rely on context managers/fixtures that do it for you, to avoid leftover data directories.
• Prefernode.safe_psql() for lightweight assertions that should fail fast; usenode.execute() when you need structured Python results.

Ildar Musin
Dmitry Ivanov
Ildus Kurbangaliev
Yury Zhuravlev