External Server Integration

def postgresql_noproc(
    host: Optional[str] = None,
    port: Union[str, int, None] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    dbname: Optional[str] = None,
    options: str = "",
    load: Optional[List[Union[Callable, str, Path]]] = None,
) -> Callable[[FixtureRequest], Iterator[NoopExecutor]]:
    """
    Create a no-process postgresql fixture factory for existing servers.

    Parameters:
    - host: PostgreSQL server hostname (default: 127.0.0.1)
    - port: PostgreSQL server port (default: 5432)
    - user: PostgreSQL username (default: postgres)
    - password: PostgreSQL password (default: None)
    - dbname: Database name (default: tests)
    - options: PostgreSQL connection options (default: "")
    - load: List of SQL files, Python callables, or import strings for initialization

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

class NoopExecutor:
    """
    Nooperator executor for existing PostgreSQL servers.
    
    Provides connection details and version detection without managing
    the server process lifecycle.
    """
    
    def __init__(
        self,
        host: str,
        port: Union[str, int],
        user: str,
        options: str,
        dbname: str,
        password: Optional[str] = None,
    ): ...
    
    @property
    def version(self) -> Any: ...
    @property
    def template_dbname(self) -> str: ...
    
    # Connection properties
    host: str
    port: int
    user: str
    password: Optional[str]
    dbname: str
    options: str

def xdistify_dbname(dbname: str) -> str:
    """
    Modify database name for pytest-xdist parallel execution.
    
    Appends worker ID to database name when running with pytest-xdist
    to prevent conflicts between parallel test workers.
    
    Parameters:
    - dbname: Base database name
    
    Returns:
    Modified database name with worker suffix if running under xdist
    """

from pytest_postgresql import factories

# Connect to local PostgreSQL server
postgresql_noproc = factories.postgresql_noproc()

def test_external_postgresql(postgresql_noproc):
    """Test connection to external PostgreSQL server."""
    assert postgresql_noproc.host == '127.0.0.1'
    assert postgresql_noproc.port == 5432
    assert postgresql_noproc.version is not None

from pytest_postgresql import factories

# Connect to PostgreSQL running in Docker container
postgresql_docker = factories.postgresql_noproc(
    host='localhost',
    port=5432,
    user='testuser',
    password='testpass',
    dbname='testdb'
)

def test_docker_postgresql(postgresql_docker):
    """Test connection to PostgreSQL in Docker."""
    assert postgresql_docker.host == 'localhost'
    assert postgresql_docker.user == 'testuser'

from pytest_postgresql import factories

# Connect to cloud PostgreSQL instance
postgresql_cloud = factories.postgresql_noproc(
    host='mydb.amazonaws.com',
    port=5432,
    user='dbuser',
    password='secure_password',
    dbname='production_test',
    options='sslmode=require'
)

def test_cloud_postgresql(postgresql_cloud):
    """Test connection to cloud PostgreSQL."""
    assert 'sslmode=require' in postgresql_cloud.options

import os
from pytest_postgresql import factories

# Use environment variables for CI/CD
postgresql_ci = factories.postgresql_noproc(
    host=os.getenv('POSTGRES_HOST', 'localhost'),
    port=int(os.getenv('POSTGRES_PORT', '5432')),
    user=os.getenv('POSTGRES_USER', 'postgres'),
    password=os.getenv('POSTGRES_PASSWORD'),
    dbname=os.getenv('POSTGRES_DB', 'test')
)

def test_ci_postgresql(postgresql_ci):
    """Test PostgreSQL in CI environment."""
    # Connection details from environment
    pass

from pytest_postgresql import factories
from pathlib import Path

def setup_test_data(**kwargs):
    """Setup test data in external server."""
    import psycopg
    with psycopg.connect(**kwargs) as conn:
        with conn.cursor() as cur:
            cur.execute("CREATE SCHEMA IF NOT EXISTS test_schema;")
            cur.execute("""
                CREATE TABLE IF NOT EXISTS test_schema.products (
                    id SERIAL PRIMARY KEY,
                    name VARCHAR(100),
                    price DECIMAL(10,2)
                );
            """)
            cur.execute("""
                INSERT INTO test_schema.products (name, price) 
                VALUES ('Test Product', 99.99)
                ON CONFLICT DO NOTHING;
            """)
        conn.commit()

postgresql_with_data = factories.postgresql_noproc(
    host='existing-server.com',
    user='app_user',
    password='app_password',
    dbname='app_test',
    load=[setup_test_data]
)

def test_external_with_data(postgresql_with_data):
    """Test external PostgreSQL with test data loaded."""
    # Test data is loaded into existing server
    pass

from pytest_postgresql import factories

# Primary database server
postgresql_primary = factories.postgresql_noproc(
    host='primary.db.com',
    port=5432,
    dbname='primary_db'
)

# Read replica server
postgresql_replica = factories.postgresql_noproc(
    host='replica.db.com',
    port=5432,
    dbname='primary_db'
)

def test_primary_replica_setup(postgresql_primary, postgresql_replica):
    """Test primary-replica database setup."""
    assert postgresql_primary.host != postgresql_replica.host
    assert postgresql_primary.dbname == postgresql_replica.dbname

from pytest_postgresql import factories
from pytest_postgresql.factories.noprocess import xdistify_dbname

# Database name will be modified for parallel workers
postgresql_parallel = factories.postgresql_noproc(
    host='shared-server.com',
    dbname='test_db'  # Becomes test_db_gw0, test_db_gw1, etc. with xdist
)

def test_parallel_execution(postgresql_parallel):
    """Test parallel execution with xdist."""
    # Each worker gets its own database name
    db_name = xdistify_dbname('test_db')
    # db_name will include worker ID if running under xdist

# SSL configuration for secure connections
postgresql_secure = factories.postgresql_noproc(
    host='secure-db.com',
    port=5432,
    user='secure_user',
    password='secure_password',
    options='sslmode=require sslcert=/path/to/client.crt sslkey=/path/to/client.key'
)

# Connection with pooling options
postgresql_pooled = factories.postgresql_noproc(
    host='pooled-db.com',
    options='application_name=pytest_app connect_timeout=10'
)

def test_version_detection(postgresql_noproc):
    """Test PostgreSQL version detection."""
    version = postgresql_noproc.version
    assert version is not None
    print(f"Connected to PostgreSQL version: {version}")