Process Management

def postgresql_proc(
    executable: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[PortType] = -1,
    user: Optional[str] = None,
    password: Optional[str] = None,
    dbname: Optional[str] = None,
    options: str = "",
    startparams: Optional[str] = None,
    unixsocketdir: Optional[str] = None,
    postgres_options: Optional[str] = None,
    load: Optional[List[Union[Callable, str, Path]]] = None,
) -> Callable[[FixtureRequest, TempPathFactory], Iterator[PostgreSQLExecutor]]:
    """
    Create a postgresql process fixture factory.

    Parameters:
    - executable: Path to pg_ctl executable (default: /usr/lib/postgresql/13/bin/pg_ctl)
    - host: Host address for connections (default: 127.0.0.1)
    - port: Port configuration - exact port, random, range, or set (default: -1 for auto-detect)
    - user: PostgreSQL username (default: postgres)
    - password: PostgreSQL password (default: None)
    - dbname: Default database name (default: tests)
    - options: PostgreSQL connection options (default: "")
    - startparams: PostgreSQL server start parameters (default: "-w")
    - unixsocketdir: Unix socket directory (default: temp directory)
    - postgres_options: Additional postgres executable options (default: "")
    - load: List of SQL files, Python callables, or import strings for initialization

    Returns:
    Function that creates PostgreSQLExecutor fixture with session scope
    """

class PostgreSQLExecutor(TCPExecutor):
    """
    PostgreSQL executor running on pg_ctl.
    
    Manages PostgreSQL server process including data directory initialization,
    server startup/shutdown, and connection management.
    """
    
    def __init__(
        self,
        executable: str,
        host: str,
        port: Union[str, int],
        user: str,
        options: str,
        startparams: str,
        unixsocketdir: str,
        dbname: str,
        postgres_options: str,
        password: Optional[str] = None,
    ): ...
    
    def start(self) -> "PostgreSQLExecutor": ...
    def stop(self) -> "PostgreSQLExecutor": ...
    def init_directory(self) -> None: ...
    def clean_directory(self) -> None: ...
    def wait_for_postgres(self) -> None: ...
    def running(self) -> bool: ...
    
    @property
    def version(self) -> Any: ...
    @property
    def template_dbname(self) -> str: ...

PortType = port_for.PortType
# Supports: exact port (int), random (None), port range (tuple), or port set (list)

from pytest_postgresql import factories

# Create basic process fixture
postgresql_proc = factories.postgresql_proc()

def test_postgresql_server(postgresql_proc):
    """Test that PostgreSQL server is running."""
    assert postgresql_proc.running() is True
    assert postgresql_proc.version is not None

from pytest_postgresql import factories

# Custom configuration with specific port and user
custom_postgresql_proc = factories.postgresql_proc(
    port=5433,
    user='test_user',
    dbname='test_db',
    unixsocketdir='/tmp/postgresql'
)

def test_custom_postgresql(custom_postgresql_proc):
    """Test custom PostgreSQL configuration."""
    assert custom_postgresql_proc.port == 5433
    assert custom_postgresql_proc.user == 'test_user'
    assert custom_postgresql_proc.dbname == 'test_db'

from pytest_postgresql import factories

# Different PostgreSQL instances for different test scenarios
postgresql_proc_main = factories.postgresql_proc(port=5432)
postgresql_proc_secondary = factories.postgresql_proc(port=5433)

def test_multiple_postgresql_instances(postgresql_proc_main, postgresql_proc_secondary):
    """Test multiple PostgreSQL instances running simultaneously."""
    assert postgresql_proc_main.running() is True
    assert postgresql_proc_secondary.running() is True
    assert postgresql_proc_main.port != postgresql_proc_secondary.port

from pytest_postgresql import factories
from pathlib import Path

def init_test_schema(**kwargs):
    """Initialize test schema."""
    import psycopg
    with psycopg.connect(**kwargs) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE users (
                    id SERIAL PRIMARY KEY,
                    name VARCHAR(100) NOT NULL,
                    email VARCHAR(100) UNIQUE
                );
            """)
        conn.commit()

postgresql_with_schema = factories.postgresql_proc(
    load=[
        Path('/path/to/schema.sql'),
        init_test_schema,
        'myapp.fixtures.load_test_data'
    ]
)

def test_preloaded_database(postgresql_with_schema):
    """Test PostgreSQL with preloaded schema."""
    # Schema and data are already loaded
    pass

def test_process_lifecycle(postgresql_proc):
    """Test PostgreSQL process lifecycle management."""
    # Server should be running
    assert postgresql_proc.running() is True
    
    # Stop the server
    postgresql_proc.stop()
    assert postgresql_proc.running() is False
    
    # Restart the server
    postgresql_proc.start()
    assert postgresql_proc.running() is True
    
    # Clean up data directory (use with caution)
    postgresql_proc.stop()
    postgresql_proc.clean_directory()

from pytest_postgresql.exceptions import ExecutableMissingException, PostgreSQLUnsupported

def test_error_handling():
    """Test error handling scenarios."""
    try:
        # This might fail if pg_ctl is not found
        postgresql_proc = factories.postgresql_proc(
            executable='/nonexistent/pg_ctl'
        )
    except ExecutableMissingException:
        # Handle missing PostgreSQL installation
        pass
    
    try:
        # This might fail with very old PostgreSQL versions
        postgresql_proc = factories.postgresql_proc()
    except PostgreSQLUnsupported:
        # Handle unsupported PostgreSQL version
        pass