test_chain_ordering() — langchain Function Reference
Architecture documentation for the test_chain_ordering() function in test_runnable_events_v1.py from the langchain codebase.
Entity Profile
Dependency Diagram
graph TD fa4fc5d4_750c_9dac_3690_f69ec02f9886["test_chain_ordering()"] 8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"] fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0 a14aef27_93bc_0ce1_f208_f5841ae9507f["_with_nulled_run_id()"] fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|calls| a14aef27_93bc_0ce1_f208_f5841ae9507f 6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"] fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5 style fa4fc5d4_750c_9dac_3690_f69ec02f9886 fill:#6366f1,stroke:#818cf8,color:#fff
Relationship Graph
Source Code
libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1528–1638
async def test_chain_ordering() -> None:
"""Test the event stream with a tool."""
def foo(a: str) -> str:
return a
def bar(a: str) -> str:
return a
chain = RunnableLambda(foo) | RunnableLambda(bar)
iterable = chain.astream_events("q", version="v1")
events = []
try:
for _ in range(10):
next_chunk = await anext(iterable)
events.append(next_chunk)
except Exception:
pass
events = _with_nulled_run_id(events)
for event in events:
event["tags"] = sorted(event["tags"])
_assert_events_equal_allow_superset_metadata(
events,
[
{
"data": {"input": "q"},
"event": "on_chain_start",
"metadata": {},
"name": "RunnableSequence",
"run_id": "",
"parent_ids": [],
"tags": [],
},
{
"data": {},
"event": "on_chain_start",
"metadata": {},
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
},
{
"data": {"chunk": "q"},
"event": "on_chain_stream",
"metadata": {},
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
},
{
"data": {"input": "q", "output": "q"},
"event": "on_chain_end",
"metadata": {},
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
},
{
"data": {},
"event": "on_chain_start",
"metadata": {},
"name": "bar",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:2"],
},
{
"data": {"chunk": "q"},
"event": "on_chain_stream",
"metadata": {},
"name": "bar",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:2"],
Domain
Subdomains
Source
Frequently Asked Questions
What does test_chain_ordering() do?
test_chain_ordering() is a function in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py.
Where is test_chain_ordering() defined?
test_chain_ordering() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py at line 1528.
What does test_chain_ordering() call?
test_chain_ordering() calls 2 function(s): _assert_events_equal_allow_superset_metadata, _with_nulled_run_id.
Analyze Your Own Codebase
Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.
Try Supermodel Free