test_event_stream_exception() — langchain Function Reference

Architecture documentation for the test_event_stream_exception() function in test_runnable_events_v2.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_event_stream_exception() — langchain Function Reference

Dependency Diagram

graph TD
  0fcf5aac_46f2_7e6b_9d87_05f7f9250002["test_event_stream_exception()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  0fcf5aac_46f2_7e6b_9d87_05f7f9250002 -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"]
  0fcf5aac_46f2_7e6b_9d87_05f7f9250002 -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655
  style 0fcf5aac_46f2_7e6b_9d87_05f7f9250002 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 360–373

async def test_event_stream_exception() -> None:
    def step(name: str, err: str | None, val: str) -> str:
        if err:
            raise ValueError(err)
        return val + name[-1]

    chain = (
        RunnableLambda(partial(step, "step1", None))
        | RunnableLambda(partial(step, "step2", "ERR"))
        | RunnableLambda(partial(step, "step3", None))
    )

    with pytest.raises(ValueError, match="ERR"):
        await _collect_events(chain.astream_events("X", version="v2"))