test_index_delete_full_recovery_after_deletion_failure() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 368–466

def test_index_delete_full_recovery_after_deletion_failure(
    record_manager: InMemoryRecordManager, vector_store: InMemoryVectorStore
) -> None:
    """Indexing some content to confirm it gets added only once."""
    loader = ToyLoader(
        documents=[
            Document(
                page_content="This is a test document.",
            ),
            Document(
                page_content="This is another document.",
            ),
        ]
    )

    with patch.object(
        record_manager,
        "get_time",
        return_value=datetime(2021, 1, 1, tzinfo=timezone.utc).timestamp(),
    ):
        assert index(
            loader,
            record_manager,
            vector_store,
            cleanup="full",
            key_encoder="sha256",
        ) == {
            "num_added": 2,
            "num_deleted": 0,
            "num_skipped": 0,
            "num_updated": 0,
        }

    loader = ToyLoader(
        documents=[
            Document(
                page_content="mutated document 1",
            ),
            Document(
                page_content="This is another document.",  # <-- Same as original
            ),
        ]
    )

    with (
        patch.object(
            record_manager,
            "get_time",
            return_value=datetime(2021, 1, 2, tzinfo=timezone.utc).timestamp(),
        ),
        patch.object(vector_store, "delete", return_value=False),
        pytest.raises(IndexingException),
    ):
        indexing_result = index(
            loader,
            record_manager,
            vector_store,
            cleanup="full",
            key_encoder="sha256",
        )

    # At this point, there should be 3 records in both the record manager
    # and the vector store
    doc_texts = {
        # Ignoring type since doc should be in the store and not a None
        vector_store.get_by_ids([uid])[0].page_content
        for uid in vector_store.store
    }
    assert doc_texts == {
        "This is a test document.",
        "mutated document 1",
        "This is another document.",
    }

    with patch.object(
        record_manager,
        "get_time",
        return_value=datetime(2021, 1, 3, tzinfo=timezone.utc).timestamp(),
    ):
        indexing_result = index(
            loader,