pgsql-test 0.1.0

PostgreSQL testing framework for Python - instant, isolated databases with automatic transaction rollback

pgsql-test

The Python counterpart to pgsql-test on npm. Instant, isolated PostgreSQL databases for each test — with automatic transaction rollbacks, context switching, and clean seeding.

New to pgpm? Check out the Workspace Setup Guide for a complete walkthrough of creating a pgpm workspace with Python tests.

Features

• Instant test DBs — each one seeded, isolated, and UUID-named
• Per-test rollback — every test runs in its own transaction with savepoint-based rollback via before_each()/after_each()
• RLS-friendly — test with role-based auth via set_context()
• pgpm integration — run database migrations using pgpm (PostgreSQL Package Manager)
• Flexible seeding — run .sql files, programmatic seeds, pgpm modules, or combine multiple strategies
• Auto teardown — no residue, no reboots, just clean exits

Installation

# Using Poetry (recommended)
poetry add pgsql-test

# Using pip
pip install pgsql-test

Quick Start

import pytest
from pgsql_test import get_connections, seed

# Basic usage
def test_basic_query():
    conn = get_connections()
    result = conn.db.query('SELECT 1 as value')
    assert result.rows[0]['value'] == 1
    conn.teardown()

# With pytest fixture
@pytest.fixture
def db():
    conn = get_connections()
    yield conn.db
    conn.teardown()

def test_with_fixture(db):
    result = db.query('SELECT 1 as value')
    assert result.rows[0]['value'] == 1

pgpm Integration

The primary use case for pgsql-test is testing PostgreSQL modules managed by pgpm. The seed.pgpm() adapter runs pgpm deploy to apply your migrations to an isolated test database.

Prerequisites

Install pgpm globally:

npm install -g pgpm

Basic pgpm Usage

import pytest
from pgsql_test import get_connections, seed

@pytest.fixture
def db():
    conn = get_connections(
        seed_adapters=[
            seed.pgpm(
                module_path="./packages/my-module",
                package="my-module"
            )
        ]
    )
    db = conn.db
    db.before_each()
    yield db
    db.after_each()
    conn.teardown()

def test_my_function(db):
    # Your pgpm module's functions are now available
    result = db.one("SELECT my_schema.my_function() as result")
    assert result['result'] == expected_value

pgpm with Dependencies

If your module depends on other pgpm packages (like @pgpm/faker), install them first:

cd packages/my-module
pgpm install @pgpm/faker

Then test:

def test_faker_integration(db):
    # @pgpm/faker functions are available after pgpm deploy
    result = db.one("SELECT faker.city('MI') as city")
    assert result['city'] is not None

pgpm Workspace Structure

A typical pgpm workspace for testing looks like:

my-workspace/
  pgpm.json                    # Workspace config
  packages/
    my-module/
      package.json             # Module metadata
      my-module.control        # PostgreSQL extension control
      pgpm.plan                # Migration plan
      deploy/
        schemas/
          my_schema.sql        # CREATE SCHEMA my_schema;
        functions/
          my_function.sql      # CREATE FUNCTION ...
      revert/
        schemas/
          my_schema.sql        # DROP SCHEMA my_schema;
      verify/
        schemas/
          my_schema.sql        # SELECT 1 FROM ...

seed.pgpm() Parameters

Parameter	Type	Description
module_path	str	Path to the pgpm module directory
package	str	Package name to deploy (required to avoid interactive prompts)
deploy_args	list[str]	Additional arguments to pass to pgpm deploy
cache	bool	Enable caching (not yet implemented)

SQL File Seeding

For simpler use cases without pgpm, seed directly from SQL files:

@pytest.fixture
def seeded_db():
    conn = get_connections(
        seed_adapters=[seed.sqlfile(['schema.sql', 'fixtures.sql'])]
    )
    yield conn.db
    conn.teardown()

def test_with_seeding(seeded_db):
    users = seeded_db.many('SELECT * FROM users')
    assert len(users) > 0

Per-Test Rollback

The before_each() and after_each() methods provide automatic transaction rollback for each test. This ensures complete isolation between tests - any changes made during a test are automatically rolled back, so each test starts with a clean slate.

How It Works

1. before_each() begins a transaction and creates a savepoint
2. Your test runs and makes changes to the database
3. after_each() rolls back to the savepoint, undoing all changes
4. The next test starts fresh with only the seeded data

Basic Pattern

@pytest.fixture
def db():
    conn = get_connections(
        seed_adapters=[seed.sqlfile(['schema.sql'])]
    )
    db = conn.db
    db.before_each()  # Begin transaction + savepoint
    yield db
    db.after_each()   # Rollback to savepoint
    conn.teardown()

def test_insert_user(db):
    # This insert will be rolled back after the test
    db.execute("INSERT INTO users (name) VALUES ('Test User')")
    result = db.one("SELECT * FROM users WHERE name = 'Test User'")
    assert result['name'] == 'Test User'

def test_user_count(db):
    # Previous test's insert is not visible here
    result = db.one("SELECT COUNT(*) as count FROM users")
    assert result['count'] == 0  # Only seeded data

Why This Matters

Without per-test rollback, tests can interfere with each other:

• Test A inserts a user
• Test B expects 0 users but finds 1
• Tests become order-dependent and flaky

With before_each()/after_each(), each test is completely isolated, making your test suite reliable and deterministic.

RLS Testing

Test Row Level Security policies by switching contexts:

def test_rls_policy(db):
    db.before_each()
    
    # Set the user context
    db.set_context({'app.user_id': '123'})
    
    # Now queries will be filtered by RLS policies
    result = db.many('SELECT * FROM user_data')
    
    db.after_each()

Seeding Strategies

pgpm Modules

seed.pgpm(module_path="./packages/my-module", package="my-module")

SQL Files

seed.sqlfile(['schema.sql', 'fixtures.sql'])

Custom Functions

seed.fn(lambda ctx: ctx['pg'].execute(
    "INSERT INTO users (name) VALUES (%s)", ('Alice',)
))

Composed Seeding

seed.compose([
    seed.pgpm(module_path="./packages/my-module", package="my-module"),
    seed.sqlfile(['fixtures.sql']),
    seed.fn(lambda ctx: ctx['pg'].execute("INSERT INTO ...")),
])

Configuration

Configure via environment variables:

export PGHOST=localhost
export PGPORT=5432
export PGUSER=postgres
export PGPASSWORD=your_password

Or pass configuration directly:

conn = get_connections(
    pg_config={
        'host': 'localhost',
        'port': 5432,
        'user': 'postgres',
        'password': 'your_password',
    }
)

API Reference

get_connections(pg_config?, connection_options?, seed_adapters?)

Creates a new isolated test database and returns connection objects.

Returns a ConnectionResult with:

• pg: PgTestClient connected as superuser
• db: PgTestClient for testing (same as pg for now)
• admin: DbAdmin for database management
• manager: PgTestConnector managing connections
• teardown(): Function to clean up

PgTestClient

• query(sql, params?): Execute SQL and return QueryResult
• one(sql, params?): Return exactly one row
• one_or_none(sql, params?): Return one row or None
• many(sql, params?): Return multiple rows
• many_or_none(sql, params?): Return rows (may be empty)
• execute(sql, params?): Execute and return affected row count
• before_each(): Start test isolation (transaction + savepoint)
• after_each(): End test isolation (rollback)
• set_context(dict): Set session variables for RLS testing

GitHub Actions Example

Here's a complete CI workflow for testing pgpm modules:

name: Test

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:17
        env:
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: password
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    env:
      PGHOST: localhost
      PGPORT: 5432
      PGUSER: postgres
      PGPASSWORD: password

    steps:
      - uses: actions/checkout@v4
      
      - uses: actions/setup-node@v4
        with:
          node-version: '20'
      
      - name: Install pgpm
        run: npm install -g pgpm
      
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install Poetry
        uses: snok/install-poetry@v1
      
      - name: Install dependencies
        run: poetry install
      
      - name: Bootstrap pgpm roles
        run: |
          pgpm admin-users bootstrap --yes
          pgpm admin-users add --test --yes
      
      - name: Run tests
        run: poetry run pytest -v

Development

# Install dependencies
poetry install

# Run tests
poetry run pytest

# Run linting
poetry run ruff check .

# Run type checking
poetry run mypy src

Related Projects

• pgsql-test - The original TypeScript/Node.js version
• pgpm - PostgreSQL Package Manager

License

MIT