test_disable_streaming() — langchain Function Reference

Architecture documentation for the test_disable_streaming() function in test_base.py from the langchain codebase.

Function python CoreAbstractions MessageSchema

Entity Profile

CoreAbstractions→ MessageSchema→ test_disable_streaming() — langchain Function Reference

Dependency Diagram

graph TD
  0737c6e2_34eb_9d3e_5cdc_b2b689ef4f2b["test_disable_streaming()"]
  8830054d_ac1e_daa9_c6c5_ff55b10d0bf3["test_base.py"]
  0737c6e2_34eb_9d3e_5cdc_b2b689ef4f2b -->|defined in| 8830054d_ac1e_daa9_c6c5_ff55b10d0bf3
  style 0737c6e2_34eb_9d3e_5cdc_b2b689ef4f2b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_base.py lines 383–404

def test_disable_streaming(
    *,
    disable_streaming: bool | Literal["tool_calling"],
) -> None:
    model = StreamingModel(disable_streaming=disable_streaming)
    assert model.invoke([]).content == "invoke"

    expected = "invoke" if disable_streaming is True else "stream"
    assert next(model.stream([])).content == expected
    assert (
        model.invoke([], config={"callbacks": [LogStreamCallbackHandler()]}).content
        == expected
    )

    expected = "invoke" if disable_streaming in {"tool_calling", True} else "stream"
    assert next(model.stream([], tools=[{"type": "function"}])).content == expected
    assert (
        model.invoke(
            [], config={"callbacks": [LogStreamCallbackHandler()]}, tools=[{}]
        ).content
        == expected
    )