test_async_callback_manager() — langchain Function Reference

Architecture documentation for the test_async_callback_manager() function in test_dispatch_custom_event.py from the langchain codebase.

Function python CoreAbstractions Serialization

Entity Profile

CoreAbstractions→ Serialization→ test_async_callback_manager() — langchain Function Reference

Dependency Diagram

graph TD
  8e9c75e7_e357_d6b9_7ca6_e64aabe5c219["test_async_callback_manager()"]
  f4cde4a2_4706_d7d1_26f9_82085d97a270["test_dispatch_custom_event.py"]
  8e9c75e7_e357_d6b9_7ca6_e64aabe5c219 -->|defined in| f4cde4a2_4706_d7d1_26f9_82085d97a270
  style 8e9c75e7_e357_d6b9_7ca6_e64aabe5c219 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/callbacks/test_dispatch_custom_event.py lines 91–111

async def test_async_callback_manager() -> None:
    """Test async callback manager."""
    callback = AsyncCustomCallbackHandler()

    run_id = uuid.UUID(int=7)

    @RunnableLambda
    async def foo(x: int, config: RunnableConfig) -> int:
        await adispatch_custom_event("event1", {"x": x}, config=config)
        await adispatch_custom_event("event2", {"x": x}, config=config)
        return x

    await foo.ainvoke(
        1,
        {"callbacks": [callback], "run_id": run_id},
    )

    assert callback.events == [
        ("event1", {"x": 1}, UUID("00000000-0000-0000-0000-000000000007"), [], {}),
        ("event2", {"x": 1}, UUID("00000000-0000-0000-0000-000000000007"), [], {}),
    ]