testgres.api¶

Testing framework for PostgreSQL and its extensions

This module was created under influence of Postgres TAP test feature (PostgresNode.pm module). It can manage Postgres clusters: initialize, edit configuration files, start/stop cluster, execute queries. The typical flow may look like:

>>> with get_new_node() as node:
...     node.init().start()
...     result = node.safe_psql('postgres', 'select 1')
...     print(result.decode('utf-8').strip())
...     node.stop()
PostgresNode(name='...', port=..., base_dir='...')
1
PostgresNode(name='...', port=..., base_dir='...')

Or:

>>> with get_new_node() as master:
...     master.init().start()
...     with master.backup() as backup:
...         with backup.spawn_replica() as replica:
...             replica = replica.start()
...             master.execute('postgres', 'create table test (val int4)')
...             master.execute('postgres', 'insert into test values (0), (1), (2)')
...             replica.catchup()  # wait until changes are visible
...             print(replica.execute('postgres', 'select count(*) from test'))
PostgresNode(name='...', port=..., base_dir='...')
[(3,)]

testgres.api.get_new_node(name=None, base_dir=None, **kwargs)¶

Simply a wrapper around PostgresNode constructor. See PostgresNode.__init__() for details.

testgres.backup¶

class testgres.backup.NodeBackup(node, base_dir=None, username=None, xlog_method=<XLogMethod.fetch: 'fetch'>)¶

Bases: object

Smart object responsible for backups

cleanup()¶

Remove all files that belong to this backup. No-op if it’s been converted to a PostgresNode (destroy=True).

log_file¶

spawn_primary(name=None, destroy=True)¶

Create a primary node from a backup.

Parameters:	name – primary’s application name. destroy – should we convert this backup into a node?
Returns:	New instance of PostgresNode.

spawn_replica(name=None, destroy=True, slot=None)¶

Create a replica of the original node from a backup.

Parameters:	name – replica’s application name. slot – create a replication slot with the specified name. destroy – should we convert this backup into a node?
Returns:	New instance of PostgresNode.

testgres.config¶

class testgres.config.GlobalConfig(**options)¶

Bases: object

Global configuration object which allows user to override default settings.

copy()¶

Return a copy of this object.

items()¶

Return setting-value pairs.

keys()¶

Return a list of all available settings.

update(config)¶

Extract setting-value pairs from ‘config’ and assign those values to corresponding settings of this GlobalConfig object.

cache_initdb = True¶

shall we use cached initdb instance?

cache_pg_config = True¶

shall we cache pg_config results?

cached_initdb_dir¶

path to a temp directory for cached initdb.

cached_initdb_unique = False¶

shall we give new node a unique system id?

error_log_lines = 20¶

N of log lines to be shown in exceptions (0=inf).

node_cleanup_full = True¶

shall we remove EVERYTHING (including logs)?

node_cleanup_on_bad_exit = False¶

remove base_dir on __exit__() via exception.

node_cleanup_on_good_exit = True¶

remove base_dir on nominal __exit__().

temp_dir¶

path to temp dir containing nodes with default ‘base_dir’.

use_python_logging = False¶

enable python logging subsystem (see logger.py).

testgres.config.configure_testgres(**options)¶

Adjust current global options. Look at the GlobalConfig to learn about existing settings.

testgres.config.pop_config()¶

Set previous GlobalConfig options from stack.

testgres.config.push_config(**options)¶

Permanently set custom GlobalConfig options and put previous settings on top of the config stack.

testgres.config.scoped_config(**options)¶

Temporarily set custom GlobalConfig options for this context. Previous options are pushed to the config stack.

Example

>>> from .api import get_new_node
>>> with scoped_config(cache_initdb=False):
...     # create a new node with fresh initdb
...     with get_new_node().init().start() as node:
...         print(node.execute('select 1'))
[(1,)]

testgres.connection¶

class testgres.connection.NodeConnection(node, dbname=None, username=None, password=None, autocommit=False)¶

Bases: object

Transaction wrapper returned by Node

begin(isolation_level=<IsolationLevel.ReadCommitted: 'read committed'>)¶

close()¶

commit()¶

connection¶

cursor¶

execute(query, *args)¶

node¶

pid¶

rollback()¶

testgres.enums¶

class testgres.enums.DumpFormat¶

Bases: enum.Enum

Available dump formats

Custom = 'custom'¶

Directory = 'directory'¶

Plain = 'plain'¶

Tar = 'tar'¶

class testgres.enums.IsolationLevel¶

Bases: enum.Enum

Transaction isolation level for NodeConnection

ReadCommitted = 'read committed'¶

ReadUncommitted = 'read uncommitted'¶

RepeatableRead = 'repeatable read'¶

Serializable = 'serializable'¶

class testgres.enums.NodeStatus¶

Bases: enum.IntEnum

Status of a PostgresNode

Running = 0¶

Stopped = 1¶

Uninitialized = 2¶

class testgres.enums.ProcessType¶

Bases: enum.Enum

Types of processes

AutovacuumLauncher = 'autovacuum launcher'¶

BackgroundWriter = 'background writer'¶

Checkpointer = 'checkpointer'¶

LogicalReplicationLauncher = 'logical replication launcher'¶

Startup = 'startup'¶

StatsCollector = 'stats collector'¶

Unknown = 'unknown'¶

WalReceiver = 'wal receiver'¶

WalSender = 'wal sender'¶

WalWriter = 'wal writer'¶

class testgres.enums.XLogMethod¶

Bases: enum.Enum

Available WAL methods for NodeBackup

fetch = 'fetch'¶

none = 'none'¶

stream = 'stream'¶

testgres.exceptions¶

exception testgres.exceptions.BackupException¶

Bases: testgres.exceptions.TestgresException

exception testgres.exceptions.CatchUpException(message=None, query=None)¶

Bases: testgres.exceptions.QueryException

exception testgres.exceptions.ExecUtilException(message=None, command=None, exit_code=0, out=None)¶

Bases: testgres.exceptions.TestgresException

exception testgres.exceptions.InitNodeException¶

Bases: testgres.exceptions.TestgresException

exception testgres.exceptions.QueryException(message=None, query=None)¶

Bases: testgres.exceptions.TestgresException

exception testgres.exceptions.StartNodeException(message=None, files=None)¶

Bases: testgres.exceptions.TestgresException

exception testgres.exceptions.TestgresException¶

Bases: Exception

exception testgres.exceptions.TimeoutException(message=None, query=None)¶

Bases: testgres.exceptions.QueryException

testgres.node¶

class testgres.node.PostgresNode(name=None, port=None, base_dir=None)¶

__init__(name=None, port=None, base_dir=None)¶

PostgresNode constructor.

Parameters:	name – node’s application name. port – port to accept connections. base_dir – path to node’s data directory.

append_conf(line='', filename='postgresql.conf', **kwargs)¶

Append line to a config file.

Parameters:	line – string to be appended to config. filename – config file (postgresql.conf by default). **kwargs – named config options.
Returns:	This instance of PostgresNode.

Examples

>>> append_conf(fsync=False)
>>> append_conf('log_connections = yes')
>>> append_conf(random_page_cost=1.5, fsync=True, ...)
>>> append_conf('postgresql.conf', 'synchronous_commit = off')

auxiliary_pids¶

Returns a dict of { ProcessType – PID }.

auxiliary_processes¶

Returns a list of auxiliary processes. Each process is represented by ProcessProxy object.

backup(**kwargs)¶

Perform pg_basebackup.

Parameters:	username – database user name. xlog_method – a method for collecting the logs (‘fetch’ | ‘stream’). base_dir – the base directory for data files and logs
Returns:	A smart object of type NodeBackup.

catchup(dbname=None, username=None)¶

Wait until async replica catches up with its master.

child_processes¶

Returns a list of all child processes. Each process is represented by ProcessProxy object.

cleanup(max_attempts=3)¶

Stop node if needed and remove its data/logs directory. NOTE: take a look at TestgresConfig.node_cleanup_full.

Parameters:	max_attempts – how many times should we try to stop()?
Returns:	This instance of PostgresNode.

connect(dbname=None, username=None, password=None, autocommit=False)¶

Connect to a database.

Parameters:	dbname – database name to connect to. username – database user name. password – user’s password. autocommit – commit each statement automatically. Also it should be set to True for statements requiring to be run outside a transaction? such as VACUUM or CREATE DATABASE.
Returns:	An instance of NodeConnection.

default_conf(fsync=False, unix_sockets=True, allow_streaming=True, allow_logical=False, log_statement='all')¶

Apply default settings to this node.

Parameters:	fsync – should this node use fsync to keep data safe? unix_sockets – should we enable UNIX sockets? allow_streaming – should this node add a hba entry for replication? allow_logical – can this node be used as a logical replication publisher? log_statement – one of (‘all’, ‘off’, ‘mod’, ‘ddl’).
Returns:	This instance of PostgresNode.

dump(filename=None, dbname=None, username=None, format=<DumpFormat.Plain: 'plain'>)¶

Dump database into a file using pg_dump. NOTE: the file is not removed automatically.

Parameters:	filename – database dump taken by pg_dump. dbname – database name to connect to. username – database user name. format – format argument plain/custom/directory/tar.
Returns:	Path to a file containing dump.

execute(query, dbname=None, username=None, password=None, commit=True)¶

Execute a query and return all rows as list.

Parameters:	query – query to be executed. dbname – database name to connect to. username – database user name. password – user’s password. commit – should we commit this query?
Returns:	A list of tuples representing rows.

free_port()¶

Reclaim port owned by this node. NOTE: does not free auto selected ports.

get_control_data()¶

Return contents of pg_control file.

init(initdb_params=None, **kwargs)¶

Perform initdb for this node.

Parameters:	initdb_params – parameters for initdb (list). fsync – should this node use fsync to keep data safe? unix_sockets – should we enable UNIX sockets? allow_streaming – should this node add a hba entry for replication?
Returns:	This instance of PostgresNode

pg_ctl(params)¶

Invoke pg_ctl with params.

Parameters:	params – arguments for pg_ctl.
Returns:	Stdout + stderr of pg_ctl.

pgbench(dbname=None, username=None, stdout=None, stderr=None, options=[])¶

Spawn a pgbench process.

Parameters:	dbname – database name to connect to. username – database user name. stdout – stdout file to be used by Popen. stderr – stderr file to be used by Popen. options – additional options for pgbench (list).
Returns:	Process created by subprocess.Popen.

pgbench_init(**kwargs)¶

Small wrapper for pgbench_run(). Sets initialize=True.

Returns:	This instance of PostgresNode.

pgbench_run(dbname=None, username=None, options=[], **kwargs)¶

Run pgbench with some options. This event is logged (see self.utils_log_file).

Parameters:	dbname – database name to connect to. username – database user name. options – additional options for pgbench (list). **kwargs – named options for pgbench. Run pgbench –help to learn more.
Returns:	Stdout produced by pgbench.

Examples

>>> pgbench_run(initialize=True, scale=2)
>>> pgbench_run(time=10)

pid¶

Return postmaster’s PID if node is running, else 0.

poll_query_until(query, dbname=None, username=None, max_attempts=0, sleep_time=1, expected=True, commit=True, suppress=None)¶

Run a query once per second until it returns ‘expected’. Query should return a single value (1 row, 1 column).

Parameters:	query – query to be executed. dbname – database name to connect to. username – database user name. max_attempts – how many times should we try? 0 == infinite sleep_time – how much should we sleep after a failure? expected – what should be returned to break the cycle? commit – should (possible) changes be committed? suppress – a collection of exceptions to be suppressed.

Examples

>>> poll_query_until('select true')
>>> poll_query_until('postgres', "select now() > '01.01.2018'")
>>> poll_query_until('select false', expected=True, max_attempts=4)
>>> poll_query_until('select 1', suppress={testgres.OperationalError})

promote(dbname=None, username=None)¶

Promote standby instance to master using pg_ctl. For PostgreSQL versions below 10 some additional actions required to ensure that instance became writable and hence dbname and username parameters may be needed.

Returns:	This instance of PostgresNode.

psql(query=None, filename=None, dbname=None, username=None, input=None, **variables)¶

Execute a query using psql.

Parameters:	query – query to be executed. filename – file with a query. dbname – database name to connect to. username – database user name. input – raw input to be passed. **variables – vars to be set before execution.
Returns:	A tuple of (code, stdout, stderr).

Examples

>>> psql('select 1')
>>> psql('postgres', 'select 2')
>>> psql(query='select 3', ON_ERROR_STOP=1)

publish(name, **kwargs)¶

Create publication for logical replication

Parameters:	pubname – publication name tables – tables names list dbname – database name where objects or interest are located username – replication username

reload(params=[])¶

Reload config files using pg_ctl.

Parameters:	params – additional arguments for pg_ctl.
Returns:	This instance of PostgresNode.

replicate(name=None, slot=None, **kwargs)¶

Create a binary replica of this node.

Parameters:	name – replica’s application name. slot – create a replication slot with the specified name. username – database user name. xlog_method – a method for collecting the logs (‘fetch’ | ‘stream’). base_dir – the base directory for data files and logs

restart(params=[])¶

Restart this node using pg_ctl.

Parameters:	params – additional arguments for pg_ctl.
Returns:	This instance of PostgresNode.

restore(filename, dbname=None, username=None)¶

Restore database from pg_dump’s file.

Parameters:	filename – database dump taken by pg_dump in custom/directory/tar formats. dbname – database name to connect to. username – database user name.

safe_psql(query=None, **kwargs)¶

Execute a query using psql.

Parameters:	query – query to be executed. filename – file with a query. dbname – database name to connect to. username – database user name. input – raw input to be passed. are passed to psql() (**kwargs) –
Returns:	psql’s output as str.

source_walsender¶

Returns master’s walsender feeding this replica.

start(params=[], wait=True)¶

Start this node using pg_ctl.

Parameters:	params – additional arguments for pg_ctl. wait – wait until operation completes.
Returns:	This instance of PostgresNode.

status()¶

Check this node’s status.

Returns:	An instance of NodeStatus.

stop(params=[], wait=True)¶

Stop this node using pg_ctl.

Parameters:	params – additional arguments for pg_ctl. wait – wait until operation completes.
Returns:	This instance of PostgresNode.

subscribe(publication, name, dbname=None, username=None, **params)¶

Create subscription for logical replication

Parameters:	name – subscription name publication – publication object obtained from publish() dbname – database name username – replication username params – subscription parameters (see documentation on CREATE SUBSCRIPTION for details)

version¶

Return PostgreSQL version for this node.

Returns:	Instance of distutils.version.LooseVersion.

class testgres.node.ProcessProxy(process, ptype=None)¶

Wrapper for psutil.Process

process¶

wrapped psutill.Process object

ptype¶

instance of ProcessType

testgres.pubsub¶

Unlike physical replication the logical replication allows users replicate only specified databases and tables. It uses publish-subscribe model with possibly multiple publishers and multiple subscribers. When initializing publisher’s node allow_logical=True should be passed to the PostgresNode.init() method to enable PostgreSQL to write extra information to the WAL needed by logical replication.

To replicate table X from node A to node B the same table structure should be defined on the subscriber’s node as logical replication don’t replicate DDL. After that publish() and subscribe() methods may be used to setup replication. Example:

>>> from testgres import get_new_node
>>> with get_new_node() as nodeA, get_new_node() as nodeB:
...     nodeA.init(allow_logical=True).start()
...     nodeB.init().start()
...
...     # create same table both on publisher and subscriber
...     create_table = 'create table test (a int, b int)'
...     nodeA.safe_psql(create_table)
...     nodeB.safe_psql(create_table)
...
...     # create publication
...     pub = nodeA.publish('mypub')
...     # create subscription
...     sub = nodeB.subscribe(pub, 'mysub')
...
...     # insert some data to the publisher's node
...     nodeA.execute('insert into test values (1, 1), (2, 2)')
...
...     # wait until changes apply on subscriber and check them
...     sub.catchup()
...
...     # read the data from subscriber's node
...     nodeB.execute('select * from test')
PostgresNode(name='...', port=..., base_dir='...')
PostgresNode(name='...', port=..., base_dir='...')
''
''
[(1, 1), (2, 2)]

class testgres.node.Publication(name, node, tables=None, dbname=None, username=None)¶

__init__(name, node, tables=None, dbname=None, username=None)¶

Constructor. Use PostgresNode.publish() instead of direct constructing publication objects.

Parameters:	name – publication name. node – publisher’s node. tables – tables list or None for all tables. dbname – database name used to connect and perform subscription. username – username used to connect to the database.

add_tables(tables, dbname=None, username=None)¶

Add tables to the publication. Cannot be used if publication was created with empty tables list.

Parameters:	tables – a list of tables to be added to the publication.

drop(dbname=None, username=None)¶

Drop publication

class testgres.node.Subscription(node, publication, name=None, dbname=None, username=None, **params)¶

__init__(node, publication, name=None, dbname=None, username=None, **params)¶

Constructor. Use PostgresNode.subscribe() instead of direct constructing subscription objects.

Parameters:	name – subscription name. node – subscriber’s node. publication – Publication object we are subscribing to (see PostgresNode.publish()). dbname – database name used to connect and perform subscription. username – username used to connect to the database. params – subscription parameters (see documentation on CREATE SUBSCRIPTION for details).

catchup(username=None)¶

Wait until subscription catches up with publication.

Parameters:	username – remote node’s user name.

disable(dbname=None, username=None)¶

Disables the running subscription.

drop(dbname=None, username=None)¶

Drops subscription

enable(dbname=None, username=None)¶

Enables the previously disabled subscription.

refresh(copy_data=True, dbname=None, username=None)¶

Disables the running subscription.