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 Runnables

Entity Profile

LangChainCore→ Runnables→ test_deduplication_v2() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain/tests/unit_tests/indexes/test_indexing.py lines 1310–1346

def test_deduplication_v2(
    record_manager: SQLRecordManager,
    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") == {
        "num_added": 3,
        "num_deleted": 0,
        "num_skipped": 1,
        "num_updated": 0,
    }

    # using in memory implementation here
    assert isinstance(vector_store, InMemoryVectorStore)
    contents = sorted(
        [document.page_content for document in vector_store.store.values()],
    )
    assert contents == ["1", "2", "3"]