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][source]¶

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¶

pgsql_cleanup_exclude_tables¶

pgsql_local¶

pgsql_local_create¶

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.