test_deduplication_v2() — langchain Function Reference

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

Function python LangChainCore ApiManagement

Entity Profile

LangChainCore→ ApiManagement→ test_deduplication_v2() — langchain Function Reference

Dependency Diagram

graph TD
  59b43df9_4dc9_61f7_68dc_3f1ae54c3eb2["test_deduplication_v2()"]
  a9fb4c74_0865_0941_ade3_563a79762cee["test_indexing.py"]
  59b43df9_4dc9_61f7_68dc_3f1ae54c3eb2 -->|defined in| a9fb4c74_0865_0941_ade3_563a79762cee
  style 59b43df9_4dc9_61f7_68dc_3f1ae54c3eb2 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 2365–2408

def test_deduplication_v2(
    record_manager: InMemoryRecordManager, vector_store: VectorStore
) -> None:
    """Check edge case when loader returns no new docs."""
    docs = [
        Document(
            page_content="1",
            metadata={"source": "1"},
        ),
        Document(
            page_content="1",
            metadata={"source": "1"},
        ),
        Document(
            page_content="2",
            metadata={"source": "2"},
        ),
        Document(
            page_content="3",
            metadata={"source": "3"},
        ),
    ]

    assert index(
        docs,
        record_manager,
        vector_store,
        cleanup="full",
        key_encoder="sha256",
    ) == {
        "num_added": 3,
        "num_deleted": 0,
        "num_skipped": 1,
        "num_updated": 0,
    }

    # using in memory implementation here
    assert isinstance(vector_store, InMemoryVectorStore)

    ids = list(vector_store.store.keys())
    contents = sorted(
        [document.page_content for document in vector_store.get_by_ids(ids)]
    )
    assert contents == ["1", "2", "3"]