test_passthrough_tap() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_passthrough_tap() — langchain Function Reference

Dependency Diagram

graph TD
  c865b13d_440d_195f_8f93_74aa1a302124["test_passthrough_tap()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  c865b13d_440d_195f_8f93_74aa1a302124 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  593bb72a_4695_6c93_b95c_e277aca006ae["batch()"]
  c865b13d_440d_195f_8f93_74aa1a302124 -->|calls| 593bb72a_4695_6c93_b95c_e277aca006ae
  fb618d44_c03b_ea8b_385b_2278dfb173d4["invoke()"]
  c865b13d_440d_195f_8f93_74aa1a302124 -->|calls| fb618d44_c03b_ea8b_385b_2278dfb173d4
  style c865b13d_440d_195f_8f93_74aa1a302124 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 1003–1067

def test_passthrough_tap(mocker: MockerFixture) -> None:
    fake = FakeRunnable()
    mock = mocker.Mock()

    seq = RunnablePassthrough[Any](mock) | fake | RunnablePassthrough[Any](mock)

    assert seq.invoke("hello", my_kwarg="value") == 5
    assert mock.call_args_list == [
        mocker.call("hello", my_kwarg="value"),
        mocker.call(5),
    ]
    mock.reset_mock()

    assert seq.batch(["hello", "byebye"], my_kwarg="value") == [5, 6]
    assert len(mock.call_args_list) == 4
    for call in [
        mocker.call("hello", my_kwarg="value"),
        mocker.call("byebye", my_kwarg="value"),
        mocker.call(5),
        mocker.call(6),
    ]:
        assert call in mock.call_args_list
    mock.reset_mock()

    assert seq.batch(["hello", "byebye"], my_kwarg="value", return_exceptions=True) == [
        5,
        6,
    ]
    assert len(mock.call_args_list) == 4
    for call in [
        mocker.call("hello", my_kwarg="value"),
        mocker.call("byebye", my_kwarg="value"),
        mocker.call(5),
        mocker.call(6),
    ]:
        assert call in mock.call_args_list
    mock.reset_mock()

    assert sorted(
        a
        for a in seq.batch_as_completed(
            ["hello", "byebye"], my_kwarg="value", return_exceptions=True
        )
    ) == [
        (0, 5),
        (1, 6),
    ]
    assert len(mock.call_args_list) == 4
    for call in [
        mocker.call("hello", my_kwarg="value"),
        mocker.call("byebye", my_kwarg="value"),
        mocker.call(5),
        mocker.call(6),
    ]:
        assert call in mock.call_args_list
    mock.reset_mock()

    assert list(
        seq.stream("hello", {"metadata": {"key": "value"}}, my_kwarg="value")
    ) == [5]
    assert mock.call_args_list == [
        mocker.call("hello", my_kwarg="value"),
        mocker.call(5),
    ]
    mock.reset_mock()