test_index_empty_doc_scoped_full() — langchain Function Reference

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

Function python CoreAbstractions RunnableInterface

Entity Profile

CoreAbstractions→ RunnableInterface→ test_index_empty_doc_scoped_full() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 1022–1104

def test_index_empty_doc_scoped_full(
    record_manager: InMemoryRecordManager, vector_store: InMemoryVectorStore
) -> None:
    """Test Indexing with scoped_full strategy."""
    loader = ToyLoader(
        documents=[
            Document(
                page_content="This is a test document.",
                metadata={"source": "1"},
            ),
            Document(
                page_content="This is another document.",
                metadata={"source": "1"},
            ),
            Document(
                page_content="This is yet another document.",
                metadata={"source": "1"},
            ),
            Document(
                page_content="This is a test document from another source.",
                metadata={"source": "2"},
            ),
        ]
    )

    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="scoped_full",
            source_id_key="source",
            key_encoder="sha256",
        ) == {
            "num_added": 4,
            "num_deleted": 0,
            "num_skipped": 0,
            "num_updated": 0,
        }

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

    loader = ToyLoader(documents=[])

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