test_index_with_upsert_kwargs() — langchain Function Reference

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

Function python CoreAbstractions RunnableInterface

Entity Profile

CoreAbstractions→ RunnableInterface→ test_index_with_upsert_kwargs() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 2761–2811

def test_index_with_upsert_kwargs(
    record_manager: InMemoryRecordManager, upserting_vector_store: InMemoryVectorStore
) -> None:
    """Test indexing with upsert_kwargs parameter."""
    mock_add_documents = MagicMock()

    with patch.object(upserting_vector_store, "add_documents", mock_add_documents):
        docs = [
            Document(
                page_content="Test document 1",
                metadata={"source": "1"},
            ),
            Document(
                page_content="Test document 2",
                metadata={"source": "2"},
            ),
        ]

        upsert_kwargs = {"vector_field": "embedding"}

        index(
            docs,
            record_manager,
            upserting_vector_store,
            upsert_kwargs=upsert_kwargs,
            key_encoder="sha256",
        )

        # Assert that add_documents was called with the correct arguments
        mock_add_documents.assert_called_once()
        call_args = mock_add_documents.call_args
        assert call_args is not None
        args, kwargs = call_args

        # Check that the documents are correct (ignoring ids)
        assert len(args[0]) == 2
        assert all(isinstance(doc, Document) for doc in args[0])
        assert [doc.page_content for doc in args[0]] == [
            "Test document 1",
            "Test document 2",
        ]
        assert [doc.metadata for doc in args[0]] == [{"source": "1"}, {"source": "2"}]

        # Check that IDs are present
        assert "ids" in kwargs
        assert isinstance(kwargs["ids"], list)
        assert len(kwargs["ids"]) == 2

        # Check other arguments
        assert kwargs["batch_size"] == 100
        assert kwargs["vector_field"] == "embedding"