test_runnable_branch_ainvoke_callbacks() — langchain Function Reference

Architecture documentation for the test_runnable_branch_ainvoke_callbacks() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_branch_ainvoke_callbacks() — langchain Function Reference

Dependency Diagram

graph TD
  fae3240c_8218_0898_ac9e_3244e1a48fd3["test_runnable_branch_ainvoke_callbacks()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  fae3240c_8218_0898_ac9e_3244e1a48fd3 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  style fae3240c_8218_0898_ac9e_3244e1a48fd3 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4630–4655

async def test_runnable_branch_ainvoke_callbacks() -> None:
    """Verify that callbacks are invoked correctly in ainvoke."""
    tracer = FakeTracer()

    async def raise_value_error(_: int) -> int:
        """Raise a value error."""
        msg = "x is too large"
        raise ValueError(msg)

    branch = RunnableBranch[int, int](
        (lambda x: x > 100, raise_value_error),
        lambda x: x - 1,
    )

    assert await branch.ainvoke(1, config={"callbacks": [tracer]}) == 0
    assert len(tracer.runs) == 1
    assert tracer.runs[0].error is None
    assert tracer.runs[0].outputs == {"output": 0}

    # Check that the chain on end is invoked
    with pytest.raises(ValueError, match="x is too large"):
        await branch.ainvoke(1000, config={"callbacks": [tracer]})

    assert len(tracer.runs) == 2
    assert "ValueError('x is too large')" in str(tracer.runs[1].error)
    assert not tracer.runs[1].outputs