testgres

PostgreSQL testing utility. Both Python 2.7 and 3.3+ are supported.

Installation

To install testgres, run:

pip install testgres

We encourage you to use virtualenv for your testing environment.

Usage

Environment

Note: by default testgres runs initdb, pg_ctl, psql provided by PATH.

There are several ways to specify a custom postgres installation:

• export PG_CONFIG environment variable pointing to the pg_config executable;
• export PG_BIN environment variable pointing to the directory with executable files.

Example:

export PG_BIN=$HOME/pg_10/bin
python my_tests.py

Logging

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.

Note: context managers (aka with) call stop() and cleanup() automatically.

testgres supports python logging, which means that you can aggregate logs from several nodes into one file:

import logging

# write everything to /tmp/testgres.log
logging.basicConfig(filename='/tmp/testgres.log')

# create two different nodes with logging
node1 = testgres.get_new_node(use_logging=True).init().start()
node2 = testgres.get_new_node(use_logging=True).init().start()

# execute a few queries
node1.execute('postgres', 'select 1')
node2.execute('postgres', 'select 2')

Examples

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

with testgres.get_new_node('test') as node:
    node.init()  # run initdb
    node.start() # start PostgreSQL
    print(node.execute('postgres', 'select 1'))
    node.stop()  # stop PostgreSQL

Let's walk through the code. First, you create a new node using:

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

or

with testgres.get_new_node('master', '/path/to/DB') as node:

where master is a node's application name. Name matters if you're testing something like replication. Function get_new_node() only creates directory structure in specified directory (or somewhere in '/tmp' if we did not specify base directory) for cluster. After that, we have to initialize the PostgreSQL cluster:

node.init()

This function runs initdb command and adds some basic configuration to postgresql.conf and pg_hba.conf files. Function init() accepts optional parameter allows_streaming which configures cluster for streaming replication (default is False). Now we are ready to start:

node.start()

Finally, our temporary cluster is able to process queries. There are four ways to run them:

Command	Description
node.psql(dbname, query)	Runs query via psql command and returns tuple (error code, stdout, stderr).
node.safe_psql(dbname, query)	Same as psql() except that it returns only stdout. If an error occures during the execution, an exception will be thrown.
node.execute(dbname, 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, username)	Returns connection wrapper (NodeConnection) capable of running several queries within a single transaction.

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()

To stop the server, run:

node.stop()

Backup & replication

It's quite easy to create a backup and start a new replica:

with testgres.get_new_node('master') as master:
    master.init().start()
    with master.backup() as backup:
        # create and start a new replica
        replica = backup.spawn_replica('replica').start()

        # catch up with master node
        replica.catchup()

        # execute a dummy query
        print(replica.execute('postgres', 'select 1'))

Benchmarks

testgres also can help you to make benchmarks using pgbench from postgres installation:

with testgres.get_new_node('master') as master:
    # start a new node
    master.init().start()

    # initialize default database for TPC-B
    master.pgbench_run(options=['-i'])

    # run benchmark for 20 seconds and show results
    print(master.pgbench_run(options=['-T', '20']))

Authors

Ildar Musin <i.musin(at)postgrespro.ru> Postgres Professional Ltd., Russia
Dmitry Ivanov <d.ivanov(at)postgrespro.ru> Postgres Professional Ltd., Russia
Ildus Kurbangaliev <i.kurbangaliev(at)postgrespro.ru> Postgres Professional Ltd., Russia
Yury Zhuravlev <stalkerg(at)gmail.com>