async def main():
    """Entrypoint to test multiple connectors (Asana, Notion, Dropbox, etc.) end-to-end."""
    # Create a test user and a unique sync ID for this sync run
    user = schemas.User(
        id=uuid.uuid4(),
        email="test@test.com",
        full_name="Test User",
        organization_id=uuid.uuid4(),
    )
    sync_id = uuid.uuid4()

    # Initialize an embedding model
    embedding_model = LocalText2Vec()

    # Create a Weaviate destination for storing chunks
    weaviate_dest = await WeaviateDestination.create(sync_id, embedding_model)

    # -------------------------------------------------------------------------
    # Test Asana Source Connector
    # -------------------------------------------------------------------------
    print("=== Testing Asana Connector ===")
    access_token = "..."
    asana_source = await AsanaSource.create(access_token)
    asana_chunks = await process_source_chunks(asana_source, weaviate_dest, buffer_size=50)
    print(f"Asana source generated {len(asana_chunks)} chunks total.\n")

    # -------------------------------------------------------------------------
    # Test Notion Source Connector
    # -------------------------------------------------------------------------
    print("=== Testing Notion Connector ===")
    notion_source = await NotionSource.create(user, sync_id)
    notion_chunks = await process_source_chunks(notion_source, weaviate_dest, buffer_size=50)
    print(f"Notion source generated {len(notion_chunks)} chunks total.\n")

    # -------------------------------------------------------------------------
    # Test Dropbox Source Connector
    # -------------------------------------------------------------------------
    print("=== Testing Dropbox Connector ===")
    from airweave.platform.sources.dropbox import DropboxSource

    dropbox_source = await DropboxSource.create(user, sync_id)
    dropbox_chunks = await process_source_chunks(dropbox_source, weaviate_dest, buffer_size=50)
    print(f"Dropbox source generated {len(dropbox_chunks)} chunks total.\n")

    # Optionally, verify chunks inserted
    await verify_chunks_in_weaviate(sync_id)

    # Optionally, add more connector-specific verifications here (like verify_asana_tasks_in_weaviate)
    # await verify_dropbox_folders_in_weaviate(sync_id)  # e.g. a function you might create
Entrypoint to test multiple connectors (Asana, Notion, Dropbox, etc.) end-to-end.
backend/local.py