tessl/pypi-pytest-postgresql

Postgresql fixtures and fixture factories for Pytest.

—

Pending

Overview

Eval results

Files

Database Connections

Name: tessl/pypi-pytest-postgresql
Author: tessl

Client fixture factory for creating database connections with automatic cleanup, transaction isolation, and database management between tests. Provides psycopg connections with comprehensive lifecycle management.

Capabilities

Client Fixture Factory

Creates database connection fixtures that automatically manage database lifecycle and cleanup.

def postgresql(
    process_fixture_name: str,
    dbname: Optional[str] = None,
    isolation_level: Optional[psycopg.IsolationLevel] = None,
) -> Callable[[FixtureRequest], Iterator[Connection]]:
    """
    Create a postgresql client fixture factory.

    Parameters:
    - process_fixture_name: Name of the process or noproc fixture to depend on
    - dbname: Override database name (default: uses process fixture's dbname)
    - isolation_level: Transaction isolation level (default: server default)

    Returns:
    Function that creates psycopg.Connection fixture with function scope
    """

Database Janitor

Manages database lifecycle including creation, population, and cleanup.

class DatabaseJanitor:
    """
    Database lifecycle manager for test isolation.
    
    Handles database creation, initialization, and cleanup to ensure
    test isolation and repeatability.
    """
    
    def __init__(
        self,
        *,
        user: str,
        host: str,
        port: Union[str, int],
        version: Union[str, float, Version],
        dbname: str,
        template_dbname: str,
        password: Optional[str] = None,
        isolation_level: Optional[psycopg.IsolationLevel] = None,
        connection_timeout: int = 60,
    ): ...
    
    def init(self) -> None: ...
    def drop(self) -> None: ...
    def load(self, load_list: List[Union[Callable, str, Path]]) -> None: ...
    def cursor(self) -> Iterator[Cursor]: ...
    def is_template(self) -> bool: ...
    
    def __enter__(self) -> "DatabaseJanitor": ...
    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException], 
        exc_tb: Optional[TracebackType],
    ) -> None: ...

Usage Examples

Basic Client Connection

from pytest_postgresql import factories

# Create process and client fixtures
postgresql_proc = factories.postgresql_proc()
postgresql = factories.postgresql('postgresql_proc')

def test_database_connection(postgresql):
    """Test basic database connection."""
    cur = postgresql.cursor()
    cur.execute("SELECT version();")
    result = cur.fetchone()
    assert result is not None
    cur.close()

Custom Database Name

from pytest_postgresql import factories

postgresql_proc = factories.postgresql_proc(dbname='main_db')
postgresql_custom = factories.postgresql('postgresql_proc', dbname='custom_test_db')

def test_custom_database(postgresql_custom, postgresql_proc):
    """Test connection to custom database."""
    cur = postgresql_custom.cursor()
    cur.execute("SELECT current_database();")
    db_name = cur.fetchone()[0]
    assert db_name == 'custom_test_db'
    cur.close()

Transaction Isolation

import psycopg
from pytest_postgresql import factories

postgresql_proc = factories.postgresql_proc()
postgresql_serializable = factories.postgresql(
    'postgresql_proc',
    isolation_level=psycopg.IsolationLevel.SERIALIZABLE
)

def test_transaction_isolation(postgresql_serializable):
    """Test transaction isolation level."""
    cur = postgresql_serializable.cursor()
    cur.execute("SHOW transaction_isolation;")
    isolation = cur.fetchone()[0]
    assert isolation == 'serializable'
    cur.close()

Multiple Database Connections

from pytest_postgresql import factories

postgresql_proc = factories.postgresql_proc()
postgresql_db1 = factories.postgresql('postgresql_proc', dbname='db1')
postgresql_db2 = factories.postgresql('postgresql_proc', dbname='db2')

def test_multiple_databases(postgresql_db1, postgresql_db2):
    """Test connections to multiple databases."""
    # Create table in first database
    cur1 = postgresql_db1.cursor()
    cur1.execute("CREATE TABLE test1 (id INT);")
    postgresql_db1.commit()
    cur1.close()
    
    # Create different table in second database
    cur2 = postgresql_db2.cursor()
    cur2.execute("CREATE TABLE test2 (id INT);")
    postgresql_db2.commit()
    cur2.close()
    
    # Verify isolation between databases
    cur1 = postgresql_db1.cursor()
    cur1.execute("SELECT tablename FROM pg_tables WHERE schemaname='public';")
    tables1 = [row[0] for row in cur1.fetchall()]
    assert 'test1' in tables1
    assert 'test2' not in tables1
    cur1.close()

Database Operations

def test_database_operations(postgresql):
    """Test various database operations."""
    cur = postgresql.cursor()
    
    # Create table
    cur.execute("""
        CREATE TABLE users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100) NOT NULL,
            email VARCHAR(100) UNIQUE,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
    """)
    
    # Insert data
    cur.execute(
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id;",
        ('John Doe', 'john@example.com')
    )
    user_id = cur.fetchone()[0]
    
    # Query data
    cur.execute("SELECT name, email FROM users WHERE id = %s;", (user_id,))
    user = cur.fetchone()
    assert user[0] == 'John Doe'
    assert user[1] == 'john@example.com'
    
    # Update data
    cur.execute(
        "UPDATE users SET name = %s WHERE id = %s;",
        ('Jane Doe', user_id)
    )
    
    # Delete data
    cur.execute("DELETE FROM users WHERE id = %s;", (user_id,))
    
    postgresql.commit()
    cur.close()

Context Manager Usage

def test_context_manager(postgresql):
    """Test using connection as context manager."""
    with postgresql:
        with postgresql.cursor() as cur:
            cur.execute("CREATE TABLE temp_table (data TEXT);")
            cur.execute("INSERT INTO temp_table (data) VALUES ('test');")
            
            cur.execute("SELECT data FROM temp_table;")
            result = cur.fetchone()
            assert result[0] == 'test'
    # Connection automatically committed/rolled back

Connection Properties

def test_connection_properties(postgresql):
    """Test connection properties and information."""
    # Connection status
    assert postgresql.status == psycopg.pq.ConnStatus.CONNECTION_OK
    
    # Server information
    info = postgresql.info
    assert info.server_version > 0
    assert info.host is not None
    assert info.port > 0
    
    # Database information
    cur = postgresql.cursor()
    cur.execute("SELECT current_database(), current_user, version();")
    db_name, user, version = cur.fetchone()
    assert db_name is not None
    assert user is not None
    assert 'PostgreSQL' in version
    cur.close()

Error Handling

import psycopg
from psycopg import errors

def test_error_handling(postgresql):
    """Test database error handling."""
    cur = postgresql.cursor()
    
    try:
        # This will fail - table doesn't exist
        cur.execute("SELECT * FROM nonexistent_table;")
    except errors.UndefinedTable as e:
        assert 'nonexistent_table' in str(e)
    
    try:
        # This will fail - syntax error
        cur.execute("INVALID SQL SYNTAX;")
    except errors.SyntaxError as e:
        assert 'syntax error' in str(e).lower()
    
    # Connection should still be usable after errors
    cur.execute("SELECT 1;")
    result = cur.fetchone()
    assert result[0] == 1
    
    cur.close()

Batch Operations

def test_batch_operations(postgresql):
    """Test batch database operations."""
    cur = postgresql.cursor()
    
    # Create table
    cur.execute("""
        CREATE TABLE products (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100),
            price DECIMAL(10,2)
        );
    """)
    
    # Batch insert
    products = [
        ('Product A', 19.99),
        ('Product B', 29.99),
        ('Product C', 39.99)
    ]
    
    cur.executemany(
        "INSERT INTO products (name, price) VALUES (%s, %s);",
        products
    )
    
    # Verify batch insert
    cur.execute("SELECT COUNT(*) FROM products;")
    count = cur.fetchone()[0]
    assert count == 3
    
    postgresql.commit()
    cur.close()

Connection Lifecycle

Automatic Cleanup

The client fixture automatically handles cleanup between tests:

After each test: Closes all open connections and drops the test database
Template database: Maintains a template database for fast test database creation
Isolation: Each test gets a fresh database instance

Manual Database Management

from pytest_postgresql.janitor import DatabaseJanitor

def test_manual_database_management(postgresql_proc):
    """Test manual database management with DatabaseJanitor."""
    janitor = DatabaseJanitor(
        user=postgresql_proc.user,
        host=postgresql_proc.host,
        port=postgresql_proc.port,
        version=postgresql_proc.version,
        dbname='manual_test_db',
        template_dbname=postgresql_proc.template_dbname,
        password=postgresql_proc.password
    )
    
    with janitor:
        # Database is created and ready
        with janitor.cursor() as cur:
            cur.execute("CREATE TABLE manual_test (id INT);")
            cur.execute("INSERT INTO manual_test (id) VALUES (1);")
            
            cur.execute("SELECT id FROM manual_test;")
            result = cur.fetchone()
            assert result[0] == 1
    # Database is automatically dropped

Install with Tessl CLI

npx tessl i tessl/pypi-pytest-postgresql

docs

configuration.md

data-loading.md

database-connections.md

external-server.md

index.md

process-management.md

tile.json

tessl/pypi-pytest-postgresql