test_async() — langchain Function Reference

Architecture documentation for the test_async() function in test_tracing_interops.py from the langchain codebase.

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_async() — langchain Function Reference

Dependency Diagram

graph TD
  353e1ca2_8a6c_ba21_9c45_8252973047c0["test_async()"]
  bf7edb6a_efe2_778a_5204_a2813723040a["TestRunnableSequenceParallelTraceNesting"]
  353e1ca2_8a6c_ba21_9c45_8252973047c0 -->|defined in| bf7edb6a_efe2_778a_5204_a2813723040a
  3719e2cc_93ad_455c_1a0f_e9a777837449["_create_parent()"]
  353e1ca2_8a6c_ba21_9c45_8252973047c0 -->|calls| 3719e2cc_93ad_455c_1a0f_e9a777837449
  500b76bd_3ec5_98ce_3a9c_3f62ed483d4a["_check_posts()"]
  353e1ca2_8a6c_ba21_9c45_8252973047c0 -->|calls| 500b76bd_3ec5_98ce_3a9c_3f62ed483d4a
  style 353e1ca2_8a6c_ba21_9c45_8252973047c0 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_tracing_interops.py lines 418–432

    async def test_async(
        self,
        method: Callable[
            [RunnableLambda, list[BaseCallbackHandler]], Coroutine[Any, Any, int]
        ],
    ) -> None:
        async def other_thing(_: int) -> AsyncGenerator[int, None]:
            yield 1

        parent = self._create_parent(other_thing)

        # Now run the chain and check the resulting posts
        assert await method(parent, [self.tracer]) == 3

        self._check_posts()