conftest.py

Overview

conftest.py is a pytest configuration and fixture file used within the InfiniFlow testing framework. Its primary role is to provide reusable test fixtures and helper functions that facilitate automated testing of document parsing and chunk management workflows in the InfiniFlow system.

This file defines:

• A custom waiting condition to ensure documents have been fully parsed before proceeding with tests.

• A pytest fixture that automates the lifecycle of document chunks: adding chunks to a document, ensuring document parsing completion, and cleaning up chunks after tests.

By encapsulating these common setup and teardown operations in conftest.py, the file enhances test modularity, reliability, and maintainability.

Detailed Explanation

Imported Modules and Functions

• time.sleep: Used to introduce deliberate wait times to mitigate race conditions or eventual consistency delays.

• pytest: Provides the testing framework and fixture decorators.

• Functions from common:

  • batch_add_chunks: Adds multiple chunks to a document.

  • delete_chunks: Deletes chunks from a document.

  • list_chunks: Retrieves chunks associated with a document.

  • list_documents: Retrieves documents from a knowledge base.

  • parse_documents: Initiates document parsing.

• wait_for from utils: A decorator utility to retry a condition function for a specified timeout.

Function: condition

@wait_for(30, 1, "Document parsing timeout")
def condition(_auth, _kb_id):
    ...

• Purpose: Serves as a polling condition to verify that all documents in a specified knowledge base (_kb_id) have completed a particular parsing run (run "3").

• Parameters:

  • _auth: Authentication credentials for API requests.

  • _kb_id: Knowledge base identifier to filter documents.

• Returns: True if all documents have run equal to "3", otherwise False.

• Usage: Decorated by @wait_for(30, 1, "Document parsing timeout"), this function is repeatedly called every 1 second for up to 30 seconds until it returns True or times out.

• Implementation Details:

  • Calls list_documents to retrieve documents related to the knowledge base.

  • Iterates through the documents checking their run status.

  • The run field indicates the parsing cycle or version; waiting for "3" ensures the latest parsing is complete.

• Example:

# Wait until all documents are parsed with run "3"
condition(auth_token, kb_id)

Fixture: add_chunks_func

@pytest.fixture(scope="function")
def add_chunks_func(request, WebApiAuth, add_document):
    ...

• Purpose: Provides a fixture for test functions that require a document pre-populated with chunks and ensures cleanup after tests.

• Scope: Function-level — a fresh fixture instance is created for each test function.

• Parameters:

  • request: pytest's built-in fixture providing test context and finalizer registration.

  • WebApiAuth: Fixture providing authentication credentials.

  • add_document: Fixture that creates and returns (kb_id, document_id) for a new document.

• Returns: A tuple (kb_id, document_id, chunk_ids):

  • kb_id: Knowledge base ID of the added document.

  • document_id: The ID of the document to which chunks were added.

  • chunk_ids: List of chunk IDs added to the document.

• Behavior:

  1. Registers a cleanup function with request.addfinalizer to delete all chunks added during the test after it finishes.

  2. Calls add_document fixture to create a new document.

  3. Invokes parse_documents with run: "1" to initiate parsing.

  4. Calls condition to wait until documents are fully parsed (run == "3").

  5. Adds 4 chunks to the created document using batch_add_chunks.

  6. Waits 1 second to ensure system state consistency (addresses issue #6487).

  7. Returns relevant IDs for test use.

• Example Usage in Tests:

def test_chunk_processing(add_chunks_func):
    kb_id, document_id, chunk_ids = add_chunks_func
    # Test logic using the document and chunks

• Cleanup Logic:

def cleanup():
    res = list_chunks(WebApiAuth, {"doc_id": document_id})
    chunk_ids = [chunk["chunk_id"] for chunk in res["data"]["chunks"]]
    delete_chunks(WebApiAuth, {"doc_id": document_id, "chunk_ids": chunk_ids})

Ensures no leftover data remains after tests, maintaining test environment isolation.

Important Implementation Details and Algorithms

• Wait-For Decorator (@wait_for):
  Used on condition to repeatedly check a condition with timeout and polling interval, improving test stability against asynchronous or delayed processing.

• Parsing Run Versioning:
  The use of run field values ("1", "3") acts as a versioning mechanism for document parsing states, allowing tests to synchronize on the completion of specific parsing cycles.

• Chunk Lifecycle Management:
  The fixture automates chunk addition and cleanup to prevent test pollution and ensure repeatability.

• Sleep Usage:
  A manual sleep(1) is employed as a workaround for known timing issues (referenced as issue #6487), highlighting practical handling of eventual consistency or asynchronous processing delays.

Interaction with Other Parts of the System

• common Module:
  Provides core API wrappers for document and chunk operations. conftest.py relies on these functions to perform actual backend interactions.

• utils.wait_for:
  Supplies the retry mechanism enabling robust waiting for asynchronous events.

• Test Suite:
  This file is automatically discovered by pytest and its fixtures are injected into other test modules, centralizing common setup and teardown logic.

• Authentication (WebApiAuth) and Document Creation (add_document) Fixtures:
  These fixtures are assumed to be defined elsewhere and provide necessary credentials and document creation capabilities.

Mermaid Diagram

The following flowchart illustrates the main functional workflow and relationships between the components and functions in this file:

flowchart TD
    A[Start: Test Function Requests add_chunks_func Fixture]
    B[add_document Fixture -> Returns kb_id, document_id]
    C[parse_documents(WebApiAuth, doc_ids=[document_id], run="1")]
    D[condition(WebApiAuth, kb_id) - Waits for Parsing Completion (run == "3")]
    E[batch_add_chunks(WebApiAuth, document_id, 4) -> Returns chunk_ids]
    F[sleep(1) - Wait for consistency]
    G[Return (kb_id, document_id, chunk_ids) to Test]
    H[After Test: Cleanup Function Called]
    I[list_chunks -> Get chunk_ids]
    J[delete_chunks -> Delete chunks]

    A --> B --> C --> D --> E --> F --> G
    G --> H --> I --> J

Summary

conftest.py acts as a foundational pytest support file that encapsulates common test setup and teardown logic related to document parsing and chunk management in InfiniFlow. Through a combination of waiting conditions, chunk lifecycle management, and cleanup routines, it ensures reliable and isolated testing of document-related features. Its integration with the common and utils modules, and pytest's fixture system, makes it a critical part of the test infrastructure.