Skip to main content

RLS TEST FIXTURE

rls-test-fixture.py

The db_as_app SAVEPOINT fixture: run RLS-policy assertions under a NOSUPERUSER NOBYPASSRLS app role, not the Testcontainers SUPERUSER bootstrap role that silently bypasses FORCE RLS.

Stark avatarStark

WHAT THIS PATTERN TEACHES

Why RLS tests under the default test SUPERUSER (BYPASSRLS=t) pass even when the policy is deleted. How to provision a runtime-matching app role and wrap the shared db fixture in SAVEPOINT + SET LOCAL ROLE + ROLLBACK so policy assertions actually fire and connection state is restored on teardown.

WHEN TO USE THIS

Every Python/asyncpg + pytest project with FORCE RLS. Use db_as_app for any test asserting a policy fires; reserve the plain db fixture for schema setup and admin-only ops. Ports to SQLAlchemy, psycopg, and Django ORM.

AT A GLANCE

@pytest_asyncio.fixture
async def db_as_app(db):
    await db.execute("SAVEPOINT rls_test")
    try:
        await db.execute(f"SET LOCAL ROLE {APP_ROLE_NAME}")
        yield db
    finally:
        await db.execute("ROLLBACK TO SAVEPOINT rls_test")

FRAMEWORK IMPLEMENTATIONS

python
@pytest_asyncio.fixture
async def db_as_app(db: asyncpg.Connection) -> AsyncIterator[asyncpg.Connection]:
    """
    Wrap the standard `db` fixture so RLS-sensitive tests run under the app
    role (BYPASSRLS=f), not the SUPERUSER bootstrap role. State is restored
    on teardown. Use the plain `db` fixture only for schema/admin operations.
    """
    await db.execute("SAVEPOINT rls_test")
    try:
        await db.execute(f"SET LOCAL ROLE {APP_ROLE_NAME}")
        # If the test sets a tenant ContextVar, wire it through:
        #   await db.execute("SELECT set_config('app.current_org_id', $1, true)", org_id)
        yield db
    finally:
        await db.execute("ROLLBACK TO SAVEPOINT rls_test")

# Provision the runtime non-owner role once per session. NOSUPERUSER and
# NOBYPASSRLS are load-bearing — without them the fixture buys you nothing.
async def provision_app_role(admin_conn: asyncpg.Connection) -> None:
    await admin_conn.execute(f"""
← All Patterns