test_abatch_iterate() — langchain Function Reference

Architecture documentation for the test_abatch_iterate() function in test_aiter.py from the langchain codebase.

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ test_abatch_iterate() — langchain Function Reference

Dependency Diagram

graph TD
  32394487_e61d_dc85_cec9_f13d61c07cfd["test_abatch_iterate()"]
  764f3343_e69e_a462_abb8_a6aab40df7f9["test_aiter.py"]
  32394487_e61d_dc85_cec9_f13d61c07cfd -->|defined in| 764f3343_e69e_a462_abb8_a6aab40df7f9
  style 32394487_e61d_dc85_cec9_f13d61c07cfd fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/utils/test_aiter.py lines 17–31

async def test_abatch_iterate(
    input_size: int, input_iterable: list[str], expected_output: list[list[str]]
) -> None:
    """Test batching function."""

    async def _to_async_iterable(iterable: list[str]) -> AsyncIterator[str]:
        for item in iterable:
            yield item

    iterator_ = abatch_iterate(input_size, _to_async_iterable(input_iterable))

    assert isinstance(iterator_, AsyncIterator)

    output = [el async for el in iterator_]
    assert output == expected_output