test_list_kbs.py

Overview

test_list_kbs.py is a comprehensive test suite designed to validate the behavior and robustness of the list_kbs function within the InfiniFlow project. The primary purpose of this file is to ensure that the knowledge base listing API behaves correctly under various conditions, including authorization checks, concurrency, pagination, sorting, and filtering.

The tests use the pytest framework and cover multiple facets:

• Authorization validation (valid/invalid tokens).

• Capability testing under concurrent requests.

• Dataset listing with various query parameters (pagination, order, filters).

• Input validation and error handling.

This file plays a critical role in maintaining the quality and reliability of the knowledge base listing feature by automating regression checks and enforcing expected behaviors.

Classes and Methods

1. TestAuthorization

Tests related to API authorization handling for the list_kbs function.

Methods

• test_auth_invalid(self, invalid_auth, expected_code, expected_message)

  • Purpose: Validates that the API correctly rejects unauthorized access attempts.

  • Parameters:

    • invalid_auth: Either None or an invalid authentication object (RAGFlowWebApiAuth instantiated with an invalid token).

    • expected_code: Expected HTTP-like error code (usually 401).

    • expected_message: Expected error message string.

  • Returns: None (assertions validate results).

  • Usage Example:

    test_auth = TestAuthorization()
    test_auth.test_auth_invalid(None, 401, "<Unauthorized '401: Unauthorized'>")
    

  • Details: This test uses pytest.mark.parametrize to cover multiple invalid authentication scenarios in one test function.

2. TestCapability

Tests the ability of the system to handle concurrent list_kbs requests.

Methods

• test_concurrent_list(self, WebApiAuth)

  • Purpose: Ensures that the list_kbs API can handle multiple simultaneous requests without failure.

  • Parameters:

    • WebApiAuth: Valid authentication object provided via pytest fixture.

  • Returns: None (assertions validate results).

  • Implementation Details:

    • Uses ThreadPoolExecutor with a maximum of 5 workers.

    • Submits 100 concurrent calls to list_kbs.

    • Waits for all futures to complete and asserts all responses have a success code (0).

  • Usage Example:

    test_cap = TestCapability()
    test_cap.test_concurrent_list(valid_auth)
    

3. TestDatasetsList

Tests focused on dataset listing functionality via the list_kbs API, including pagination, sorting, filtering, and parameter validation.

This class uses the add_datasets pytest fixture to prepare the environment with test datasets.

Methods

• test_params_unset(self, WebApiAuth)

  • Tests calling list_kbs with params=None.

  • Expects default behavior returning 5 knowledge bases.

• test_params_empty(self, WebApiAuth)

  • Tests calling list_kbs with empty dictionary {} as params.

  • Expects similar results as unset params.

• test_page(self, WebApiAuth, params, expected_page_size)

  • Parameterized test for different page and page_size values.

  • Checks the number of knowledge bases returned matches expectations.

  • Covers normal paging, last pages, pages beyond max, and string page numbers.

• test_page_invalid(self, WebApiAuth, params, expected_code, expected_message) (Skipped)

  • Tests invalid page parameter values (e.g., zero, non-integer strings).

  • Expects error codes and messages indicating invalid input.

• test_page_none(self, WebApiAuth)

  • Tests how list_kbs behaves when page is explicitly set to None.

• test_page_size(self, WebApiAuth, params, expected_page_size)

  • Parameterized test for various page_size inputs.

  • Includes testing minimum, medium, maximum, oversized, and string values.

• test_page_size_invalid(self, WebApiAuth, params, expected_code, expected_message) (Skipped)

  • Tests invalid page_size values (zero, non-integer strings).

  • Expects appropriate error codes and messages.

• test_page_size_none(self, WebApiAuth)

  • Tests behavior when page_size is set to None.

• test_orderby(self, WebApiAuth, params, assertions)

  • Tests sorting behavior using the orderby parameter.

  • Example checks sorting by update_time in descending order.

  • Uses the utility function is_sorted for validation.

• test_desc(self, WebApiAuth, params, assertions)

  • Tests the effect of desc parameter (True/False) on sorting order.

  • Validates sorted order accordingly.

• test_parser_id(self, WebApiAuth, params, expected_page_size)

  • Tests filtering by parser_id.

  • Checks the count of returned knowledge bases matches expectations.

Important Implementation Details

• Concurrency Testing: Uses Python’s concurrent.futures.ThreadPoolExecutor to simulate multiple simultaneous API requests, ensuring thread safety and performance under load.

• Parameter Validation: The tests check correct handling of both valid and invalid API parameters, including pagination, sorting, and filtering.

• Error Handling: Skipped tests (marked with @pytest.mark.skip) suggest known issues or planned future tests to handle invalid inputs gracefully.

• Fixtures: Relies on pytest fixtures such as WebApiAuth for authenticated sessions and add_datasets to populate test data, indicating integration with a larger test infrastructure.

• Utility Functions: Uses external utility is_sorted to assert order correctness of returned datasets.

Interaction with Other System Components

• list_kbs Function: The central tested function imported from the common module. It interfaces with the knowledge base API to fetch lists of knowledge bases.

• Authentication Library (libs.auth): Uses RAGFlowWebApiAuth to create authentication tokens, including invalid tokens for negative testing.

• Configuration (configs): Imports constants like INVALID_API_TOKEN to simulate unauthorized access.

• Utilities (utils): Uses helper functions such as is_sorted to verify sorting behavior.

• Test Fixtures: The tests utilize fixtures like WebApiAuth and add_datasets, which are defined elsewhere in the test suite, to set up authentication contexts and test data.

Usage Example

A typical test run is executed via pytest command-line, with markers such as -m p1 to selectively run priority 1 tests:

pytest test_list_kbs.py -m p1

Or to run all tests:

pytest test_list_kbs.py

Mermaid Class Diagram

classDiagram
    class TestAuthorization {
        +test_auth_invalid(invalid_auth, expected_code, expected_message)
    }
    class TestCapability {
        +test_concurrent_list(WebApiAuth)
    }
    class TestDatasetsList {
        +test_params_unset(WebApiAuth)
        +test_params_empty(WebApiAuth)
        +test_page(WebApiAuth, params, expected_page_size)
        +test_page_invalid(WebApiAuth, params, expected_code, expected_message)
        +test_page_none(WebApiAuth)
        +test_page_size(WebApiAuth, params, expected_page_size)
        +test_page_size_invalid(WebApiAuth, params, expected_code, expected_message)
        +test_page_size_none(WebApiAuth)
        +test_orderby(WebApiAuth, params, assertions)
        +test_desc(WebApiAuth, params, assertions)
        +test_parser_id(WebApiAuth, params, expected_page_size)
    }
    TestDatasetsList ..|> pytest.Fixture : uses
    TestCapability ..|> concurrent.futures.ThreadPoolExecutor : uses
    TestAuthorization ..|> common.list_kbs : tests
    TestCapability ..|> common.list_kbs : tests
    TestDatasetsList ..|> common.list_kbs : tests

Summary

test_list_kbs.py is a critical quality assurance file for the InfiniFlow project, focusing on validating the knowledge base listing API. It covers authorization, concurrency, parameter correctness, sorting, and filtering, ensuring that the API behaves correctly under diverse scenarios. The test cases are well organized into classes by concern and utilize pytest's advanced features such as parameterization and fixtures, integrating with other system modules like authentication, configuration, and utilities.

This file helps catch regressions early, enforce API contract, and confirm system resilience under concurrent access, contributing significantly to the stability and reliability of the knowledge base service.

End of Documentation