Wey-gu/py-pglite：PythonのPgliteラッパーのテスト用。 sqliteと同じように、postgresを使用してアプリをテストします。

import pytest
from sqlmodel import Session, SQLModel, Field, select
from py_pglite import pglite_session

# Your models
class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

# Test with automatic PGlite management
def test_user_creation(pglite_session: Session):
    user = User(name="Alice", email="[email protected]")
    pglite_session.add(user)
    pglite_session.commit()
    
    # Query back
    users = pglite_session.exec(select(User)).all()
    assert len(users) == 1
    assert users[0].name == "Alice"

from py_pglite import PGliteManager, PGliteConfig

# Custom configuration
config = PGliteConfig(
    timeout=30,
    cleanup_on_exit=True,
    log_level="DEBUG"
)

# Manual management
with PGliteManager(config) as manager:
    engine = manager.get_engine()
    SQLModel.metadata.create_all(engine)
    
    with Session(engine) as session:
        # Your database operations here
        pass

from py_pglite import PGliteConfig

config = PGliteConfig(
    timeout=30,               # Startup timeout in seconds
    cleanup_on_exit=True,     # Auto cleanup on exit
    log_level="INFO",         # Logging level (DEBUG/INFO/WARNING/ERROR)
    socket_path="/tmp/.s.PGSQL.5432",  # Custom socket path
    work_dir=None,            # Working directory (None = temp dir)
    node_modules_check=True,  # Verify node_modules exists
    auto_install_deps=True,   # Auto-install npm dependencies
)

from py_pglite import utils

# Database cleanup utilities
utils.clean_database_data(engine)                    # Clean all data
utils.clean_database_data(engine, exclude_tables=["users"])  # Exclude tables
utils.reset_sequences(engine)                        # Reset auto-increment sequences
utils.verify_database_empty(engine)                  # Check if database is empty

# Schema operations
utils.create_test_schema(engine, "test_schema")      # Create test schema
utils.drop_test_schema(engine, "test_schema")        # Drop test schema

# Get table statistics
row_counts = utils.get_table_row_counts(engine)      # Dict of table row counts

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from sqlmodel import Session
from py_pglite import pglite_engine

app = FastAPI()

def get_db():
    # Production database dependency
    pass

@app.post("/users/")
def create_user(user_data: dict, db: Session = Depends(get_db)):
    # Your endpoint logic
    pass

# Test with PGlite
def test_create_user_endpoint(pglite_engine):
    # Override database dependency
    def override_get_db():
        with Session(pglite_engine) as session:
            yield session
    
    app.dependency_overrides[get_db] = override_get_db
    
    with TestClient(app) as client:
        response = client.post("/users/", json={"name": "Bob"})
        assert response.status_code == 200

def test_complex_operations(pglite_session: Session):
    # Create related data
    user = User(name="Alice", email="[email protected]")
    pglite_session.add(user)
    pglite_session.commit()
    pglite_session.refresh(user)
    
    # Create dependent records
    orders = [
        Order(user_id=user.id, amount=100.0),
        Order(user_id=user.id, amount=250.0),
    ]
    pglite_session.add_all(orders)
    pglite_session.commit()
    
    # Complex query with joins
    result = pglite_session.exec(
        select(User.name, func.sum(Order.amount))
        .join(Order)
        .group_by(User.name)
    ).first()
    
    assert result[0] == "Alice"
    assert result[1] == 350.0

# ✅ Recommended: Multiple sessions with same engine
with PGliteManager() as manager:
    engine = manager.get_engine()
    
    # Multiple sessions work perfectly
    session1 = Session(engine)
    session2 = Session(engine)
    session3 = Session(engine)

# ❌ Not recommended: Multiple engines from same manager
with PGliteManager() as manager:
    engine1 = manager.get_engine()  # Can cause connection conflicts
    engine2 = manager.get_engine()  # when used simultaneously

# Pattern 1: Simple CRUD testing
def test_user_crud(pglite_session):
    # Create
    user = User(name="Test", email="[email protected]")
    pglite_session.add(user)
    pglite_session.commit()
    
    # Read
    found_user = pglite_session.get(User, user.id)
    assert found_user.name == "Test"
    
    # Update
    found_user.name = "Updated"
    pglite_session.commit()
    
    # Delete
    pglite_session.delete(found_user)
    pglite_session.commit()

# Pattern 2: Custom cleanup
def test_with_custom_cleanup(pglite_engine):
    SQLModel.metadata.create_all(pglite_engine)
    
    with Session(pglite_engine) as session:
        # Your test logic
        pass
    
    # Custom cleanup if needed
    utils.clean_database_data(pglite_engine)