test_aincremental_fails_with_bad_source_ids() — langchain Function Reference

Source Code

libs/langchain/tests/unit_tests/indexes/test_indexing.py lines 462–508

async def test_aincremental_fails_with_bad_source_ids(
    arecord_manager: SQLRecordManager,
    vector_store: InMemoryVectorStore,
) -> None:
    """Test indexing with incremental deletion strategy."""
    loader = ToyLoader(
        documents=[
            Document(
                page_content="This is a test document.",
                metadata={"source": "1"},
            ),
            Document(
                page_content="This is another document.",
                metadata={"source": "2"},
            ),
            Document(
                page_content="This is yet another document.",
                metadata={"source": None},
            ),
        ],
    )

    with pytest.raises(
        ValueError,
        match="Source id key is required when cleanup mode is incremental "
        "or scoped_full",
    ):
        # Should raise an error because no source id function was specified
        await aindex(
            loader,
            arecord_manager,
            vector_store,
            cleanup="incremental",
        )

    with pytest.raises(
        ValueError,
        match="Source IDs are required when cleanup mode is incremental or scoped_full",
    ):
        # Should raise an error because no source id function was specified
        await aindex(
            loader,
            arecord_manager,
            vector_store,
            cleanup="incremental",
            source_id_key="source",
        )