test_indexing_custom_batch_size() — langchain Function Reference

Architecture documentation for the test_indexing_custom_batch_size() function in test_indexing.py from the langchain codebase.

Function python CoreAbstractions RunnableInterface

Entity Profile

CoreAbstractions→ RunnableInterface→ test_indexing_custom_batch_size() — langchain Function Reference

Dependency Diagram

graph TD
  addafe0e_702a_4b60_3bf7_35be1a5fa247["test_indexing_custom_batch_size()"]
  576ad89d_c8dc_eddf_9cd2_c8ae0e7c9978["test_indexing.py"]
  addafe0e_702a_4b60_3bf7_35be1a5fa247 -->|defined in| 576ad89d_c8dc_eddf_9cd2_c8ae0e7c9978
  style addafe0e_702a_4b60_3bf7_35be1a5fa247 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 2556–2590

def test_indexing_custom_batch_size(
    record_manager: InMemoryRecordManager, vector_store: InMemoryVectorStore
) -> None:
    """Test indexing with a custom batch size."""
    docs = [
        Document(
            page_content="This is a test document.",
            metadata={"source": "1"},
        ),
    ]
    ids = [_get_document_with_hash(doc, key_encoder="sha256").id for doc in docs]

    batch_size = 1

    original = vector_store.add_documents

    try:
        mock_add_documents = MagicMock()
        vector_store.add_documents = mock_add_documents  # type: ignore[method-assign]

        index(
            docs,
            record_manager,
            vector_store,
            batch_size=batch_size,
            key_encoder="sha256",
        )
        args, kwargs = mock_add_documents.call_args
        doc_with_id = Document(
            id=ids[0], page_content="This is a test document.", metadata={"source": "1"}
        )
        assert args == ([doc_with_id],)
        assert kwargs == {"ids": ids, "batch_size": batch_size}
    finally:
        vector_store.add_documents = original  # type: ignore[method-assign]