test_on_chain_end_updates_when_not_defers_inputs() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/tracers/test_langchain.py lines 457–492

def test_on_chain_end_updates_when_not_defers_inputs() -> None:
    """Tests `_on_chain_end` calls update (PATCH) when `defers_inputs` is not set."""
    client = unittest.mock.MagicMock(spec=Client)
    client.tracing_queue = None
    tracer = LangChainTracer(client=client)

    run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a")
    tracer.on_chain_start(
        {"name": "test_chain"},
        {"input": "hello"},
        run_id=run_id,
    )

    run = tracer.run_map[str(run_id)]
    run.outputs = {"output": "result"}

    persist_called = False
    update_called = False

    def mock_persist(_: Any) -> None:
        nonlocal persist_called
        persist_called = True

    def mock_update(_: Any) -> None:
        nonlocal update_called
        update_called = True

    with (
        unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist),
        unittest.mock.patch.object(tracer, "_update_run_single", mock_update),
    ):
        tracer._on_chain_end(run)

    # Should call update (PATCH), not persist (POST) for normal inputs
    assert not persist_called
    assert update_called