PostgreSQL

In order to enable postgres support you have to add testsuite.database.pgsql_plugin to pytest_plugins list in your conftest.py file and configure postgresql schemas location.

By default testsuite starts PostgreSQL service. In this case PostgreSQL installation is required.

Currently pgsql plugin uses synchronous psycopg2 driver.

Pgsql plugin creates database schema once. And then populates database with data fixtures on each test. It looks for database fixtures by the following names:

• file pg_DBNAME.sql

• directory pg_DBNAME/

Customize port

Testsuite may start postgres with custom port, if TESTSUITE_POSTGRESQL_PORT environment variable is specified

Use external instance

You can force it to use your own postgres installation with command-line option --postgresql=postgresql://db-postgresql/.

Reuse database between simultaneous sessions

In general, testsuite does not support running simultaneous sessions, except when testsuite is started with --postgresql-keep-existing-db or --service-runner-mode flag.

In service runner mode testsuite starts the service and waits indefinitely so that developer can attach to running service with debugger.

If one session in service runner mode creates database and applies schemas, then the next one will skip applying schemas on initialization, unless schemas were modified since then.

Example integration

from testsuite.databases.pgsql import discover

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.pgsql.pytest_plugin',
]


@pytest.fixture(scope='session')
def pgsql_local(pgsql_local_create):
    tests_dir = pathlib.Path(__file__).parent
    sqldata_path = tests_dir.joinpath('../schemas/postgresql')
    databases = discover.find_schemas('service_name', [sqldata_path])
    return pgsql_local_create(list(databases.values()))

Database access example

def test_read_from_database(pgsql):
    cursor = pgsql['chat_messages'].cursor()
    cursor.execute(
        'SELECT username, text FROM messages WHERE id = %s', (data['id'],),
    )
    record = cursor.fetchone()
    assert record == ('foo', 'bar')

Environment variables

TESTSUITE_POSTGRESQL_PORT

Use to override Postgresql server port. Default is 15433.

Functions

find_schemas

testsuite.databases.pgsql.discover.find_schemas(service_name: str | None, schema_dirs: list[Path]) → dict[str, PgShardedDatabase]

Read database schemas from directories schema_dirs.

|- schema_path/
  |- database1.sql
  |- database2.sql

Parameters:

• service_name – service name used as prefix for database name if not empty, e.g. “servicename_dbname”.

• schema_dirs – list of pathes to scan for schemas

Returns:

Dict[str, PgShardedDatabase] where key is database name as stored in PgShard.dbname

Fixtures

pgsql

testsuite.databases.pgsql.pytest_plugin.pgsql()

Returns str to testsuite.databases.pgsql.control.PgDatabaseWrapper dictionary

Example usage:

def test_pg(pgsql):
    cursor = pgsql['example_db'].cursor()
    cursor.execute('SELECT ... FROM ...WHERE ...')
    assert list(cusror) == [...]

pgsql_cleanup_exclude_tables

testsuite.databases.pgsql.pytest_plugin.pgsql_cleanup_exclude_tables()

Redefine this fixture when you don’t need to clean some tables. For example postgis create table with spatial reference systems. To use postgis you need to add this fixture with spatial_ref_sys table.

@pytest.fixture(scope='session')
def pgsql_cleanup_exclude_tables():
    return frozenset({'public.spatial_ref_sys'})

pgsql_local

testsuite.databases.pgsql.pytest_plugin.pgsql_local()

Configures local pgsql instance.

Returns:

ServiceLocalConfig instance.

In order to use pgsql fixture you have to override pgsql_local() in your local conftest.py file, example:

@pytest.fixture(scope='session')
def pgsql_local(pgsql_local_create):
    databases = discover.find_schemas(
        'service_name', [PG_SCHEMAS_PATH])
    return pgsql_local_create(list(databases.values()))

Sometimes it is desirable to have tests-only database, maybe used in one particular test or tests group. This can be achieved by by overriding pgsql_local fixture in your test file:

@pytest.fixture
def pgsql_local(pgsql_local_create):
    databases = discover.find_schemas(
        'testsuite', [pathlib.Path('custom/pgsql/schema/path')])
    return pgsql_local_create(list(databases.values()))

pgsql_local provides access to PostgreSQL connection parameters:

def get_custom_connection_string(pgsql_local):
    conninfo = pgsql_local['database_name']
    custom_dsn: str = conninfo.replace(options='-c opt=val').get_dsn()
    return custom_dsn

pgsql_local_create

testsuite.databases.pgsql.pytest_plugin.pgsql_local_create(databases)

Creates pgsql configuration.

Parameters:

databases – List of databases.

Returns:

ServiceLocalConfig instance.

Marks

pytest.mark.pgsql

pytest.mark.pgsql(dbname, files=(), directories=(), queries=())

Use this mark to override specify extra data fixtures in a per-test manner.

Parameters:

• dbname – Database name.

• files – List of filenames to apply to the database.

• directories – List of directories to apply to the database.

• directories – List of queries to apply to the database.

Classes

class testsuite.databases.pgsql.pytest_plugin.ServiceLocalConfig

__getitem__(dbname: str) → PgConnectionInfo

Get testsuite.databases.pgsql.connection.PgConnectionInfo instance by database name

class testsuite.databases.pgsql.connection.PgConnectionInfo

PostgreSQL connection parameters

get_dsn() → str

PostgreSQL connection string in DSN format

get_uri() → str

PostgreSQL connection string in URI format

replace(**kwargs) → PgConnectionInfo

Return a new PgConnectionInfo value replacing specified fields with new values

class testsuite.databases.pgsql.control.PgDatabaseWrapper

property conn: connection

Returns:

psycopg2.extensions.connection

property conninfo: PgConnectionInfo

returns testsuite.databases.pgsql.connection.PgConnectionInfo

cursor(**kwargs) → cursor

Returns:

psycopg2.extensions.cursor

dict_cursor(**kwargs) → cursor

Returns dictionary cursor, see psycopg2.extras.DictCursor

Returns:

psycopg2.extensions.cursor