Function isolated_schema

hindsight-api-slim/tests/test_webhooks.py:1018–1047

Provision a fresh non-default schema with the full migration tree, then swap the memory engine's tenant extension so all subsequent operations resolve to it. Drops the schema on teardown.

(memory: MemoryEngine, pg0_db_url)

@pytest_asyncio.fixture
async def isolated_schema(memory: MemoryEngine, pg0_db_url):
    """Provision a fresh non-default schema with the full migration tree, then
    swap the memory engine's tenant extension so all subsequent operations
    resolve to it. Drops the schema on teardown.
    """
    import asyncpg

    from hindsight_api.migrations import run_migrations

    schema_name = f"tenant_wh_iso_{uuid.uuid4().hex[:8]}"

    # Run migrations to provision the schema with all tables (webhooks, banks,
    # async_operations, ...). This is the same path a real multi-tenant
    # extension would take to provision a new tenant schema.
    run_migrations(pg0_db_url, schema=schema_name)

    original_ext = memory._tenant_extension
    memory._tenant_extension = _NonDefaultSchemaTenantExtension(schema_name)

    try:
        yield schema_name
    finally:
        memory._tenant_extension = original_ext
        # Drop the test schema. Use a dedicated connection so we don't depend
        # on the pool's state.
        conn = await asyncpg.connect(pg0_db_url)
        try:
            await conn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')
        finally:
            await conn.close()


class TestWebhookSchemaIsolation: