test_incremental_indexing_with_batch_size() — langchain Function Reference

Source Code

libs/langchain/tests/unit_tests/indexes/test_indexing.py lines 799–863

def test_incremental_indexing_with_batch_size(
    record_manager: SQLRecordManager,
    vector_store: InMemoryVectorStore,
) -> None:
    """Test indexing with incremental indexing."""
    loader = ToyLoader(
        documents=[
            Document(
                page_content="1",
                metadata={"source": "1"},
            ),
            Document(
                page_content="2",
                metadata={"source": "1"},
            ),
            Document(
                page_content="3",
                metadata={"source": "1"},
            ),
            Document(
                page_content="4",
                metadata={"source": "1"},
            ),
        ],
    )

    with patch.object(
        record_manager,
        "get_time",
        return_value=_JANUARY_SECOND,
    ):
        assert index(
            loader,
            record_manager,
            vector_store,
            cleanup="incremental",
            source_id_key="source",
            batch_size=2,
        ) == {
            "num_added": 4,
            "num_deleted": 0,
            "num_skipped": 0,
            "num_updated": 0,
        }

        assert index(
            loader,
            record_manager,
            vector_store,
            cleanup="incremental",
            source_id_key="source",
            batch_size=2,
        ) == {
            "num_added": 0,
            "num_deleted": 0,
            "num_skipped": 4,
            "num_updated": 0,
        }

    doc_texts = {
        # Ignoring type since doc should be in the store and not a None
        vector_store.store.get(uid).page_content  # type: ignore[union-attr]
        for uid in vector_store.store
    }
    assert doc_texts == {"1", "2", "3", "4"}