test_list_sessions_with_chat_assistant.py
Overview
This file contains automated tests for the functionality related to listing sessions associated with chat assistants in the InfiniFlow platform. It primarily focuses on validating the API endpoint behavior that retrieves session lists filtered and paginated by various parameters, ensuring correct authorization, parameter handling, sorting, filtering, and concurrency robustness.
The tests are implemented using the pytest framework and cover both positive and negative cases, including invalid authorization, edge cases in pagination, sorting order, filtering by name and session IDs, and concurrent access scenarios.
Detailed Component Descriptions
Imports and Dependencies
concurrent.futures.ThreadPoolExecutor, as_completed
Used to perform concurrent API calls to test thread-safety and concurrency behavior.pytest
Testing framework used for structuring and running test cases.common:
delete_chat_assistantslist_session_with_chat_assistants
These are utility functions that interact with the API to delete chat assistants and list sessions respectively.
configs:
INVALID_API_TOKEN
A constant representing an invalid API token for testing authorization errors.
libs.auth:
RAGFlowHttpApiAuth
Authentication class used to create authorized API request objects.
utils:
is_sorted
Utility function to verify if a list of dictionaries is sorted by a specified key.
Classes and Methods
1. TestAuthorization
Tests related to the authorization mechanism when listing sessions with chat assistants.
Method: test_invalid_auth
Purpose:
Validates the API response when called with invalid or missing authorization credentials.Parameters:
invalid_auth: Authorization object or None to simulate missing or invalid auth.expected_code: Expected error code returned by the API.expected_message: Expected error message string.
Behavior:
Callslist_session_with_chat_assistantswith the given authorization and asserts that the response code and message match expectations.Usage Example:
test_auth = RAGFlowHttpApiAuth(INVALID_API_TOKEN) result = list_session_with_chat_assistants(test_auth, "some_chat_assistant_id") assert result["code"] == 109 assert "API key is invalid" in result["message"]
2. TestSessionsWithChatAssistantList
Contains extensive test cases for the listing sessions API, focusing on parameter validation, filtering, sorting, and concurrency.
Parameterized Tests in this Class
a. test_page
Purpose:
Validates pagination behavior using thepageandpage_sizeparameters.Parameters:
params: Dictionary containingpageandpage_sizevalues.expected_code: Expected API response code.expected_page_size: Expected number of session items in response data.expected_message: Expected error message if any.
Notes:
Some test cases are marked as skipped due to known issues or unsupported behaviors (e.g., negative or non-integer page values).Example:
Fetch page 2 with page size 2 and expect 2 results:params = {"page": 2, "page_size": 2} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert res["code"] == 0 assert len(res["data"]) == 2
b. test_page_size
Purpose:
Validates behavior of thepage_sizeparameter independently.Parameters:
Similar totest_pagebut focuses on varyingpage_size.Example:
Request page size 1 and expect exactly 1 session:params = {"page_size": 1} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert len(res["data"]) == 1
c. test_orderby
Purpose:
Tests sorting of session lists by different fields (create_time,update_time,name).Parameters:
params: Dict withorderbyand optionaldesc.expected_code: Expected API response code.assertions: Lambda to validate sorting correctness.expected_message: Expected error message for invalid sort fields.
Implementation Detail:
Usesis_sortedutility to check if the list is sorted ascending or descending as specified.Example:
Verify sessions are sorted bycreate_timedescending:params = {"orderby": "create_time", "desc": "true"} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert is_sorted(res["data"], "create_time", True)
d. test_desc
Purpose:
Validates thedesc(descending order) boolean parameter that modifies sort order.Parameters:
Similar totest_orderby, focusing ondescvariations (true,false, boolean values).Example:
Sort bycreate_timeascending:params = {"desc": "false"} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert is_sorted(res["data"], "create_time", False)
e. test_name
Purpose:
Filters sessions by theirnameattribute.Parameters:
params: Dict containingnamefilter.expected_code,expected_num,expected_message: Expected results.
Example:
Search for sessions named"session_with_chat_assistant_1":params = {"name": "session_with_chat_assistant_1"} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert res["data"][0]["name"] == "session_with_chat_assistant_1"
f. test_id
Purpose:
Filters sessions byid.Parameters:
session_id: Can be None, empty string, specific session ID, or callable to fetch a session ID dynamically.Other expected results similar to above.
Example:
Request session by specific ID:params = {"id": "some_session_id"} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert res["data"][0]["id"] == "some_session_id"
g. test_name_and_id
Purpose:
Combines filtering by bothnameandidto validate compound filtering behavior.Parameters:
Combination ofsession_idandname, with expected results.
Additional Tests
h. test_concurrent_list
Purpose:
Tests concurrent listing requests to verify thread safety and API stability under load.Implementation:
Uses aThreadPoolExecutorto spawn 100 concurrent calls tolist_session_with_chat_assistants.Assertions:
Ensures all responses return success code0.
i. test_invalid_params
Purpose:
Verifies that unknown or extraneous parameters do not cause errors and default behavior is maintained.Example:
Pass an unsupported parameter:params = {"a": "b"} res = list_session_with_chat_assistants(auth, chat_assistant_id, params) assert res["code"] == 0
j. test_list_chats_after_deleting_associated_chat_assistant
Purpose:
Validates that after deleting a chat assistant, attempts to list sessions for it return the appropriate authorization error.Behavior:
Deletes the chat assistant.
Attempts to list sessions with the deleted assistant ID.
Expects error code
102and a message indicating lack of ownership.
Implementation Details and Algorithms
Pagination and Sorting:
The API supports pagination withpageandpage_sizeparameters and sorting withorderbyanddesc. The tests ensure these parameters are handled correctly, including type conversions (e.g., string to int, string boolean to boolean), error handling for invalid inputs, and default fallbacks.Filtering:
Sessions can be filtered byidandname. The tests cover single and combined filters and check correct filtering behavior.Concurrency Testing:
Uses Python'sThreadPoolExecutorto simulate multiple simultaneous API calls, ensuring thread safety and consistent API responses under concurrent loads.Error Handling:
Tests verify that the API returns meaningful error codes and messages for invalid authorization, invalid parameters, and attempts to access deleted or unauthorized resources.
Interaction with Other System Components
API Functions:
list_session_with_chat_assistants(fromcommon): Core function tested here, responsible for making API calls to the backend to retrieve session lists.delete_chat_assistants(fromcommon): Used to delete chat assistants to test post-deletion behavior.
Authentication:
RAGFlowHttpApiAuth(fromlibs.auth): Provides authenticated API request objects for authorized access.
Configuration:
INVALID_API_TOKEN(fromconfigs): Used to simulate invalid authentication scenarios.
Utilities:
is_sorted(fromutils): Used to verify sorting correctness of API responses.
Testing Framework:
pytest: All tests are structured and run using pytest with parametric testing and test markers.
Usage Example of the Main Tested Function
from libs.auth import RAGFlowHttpApiAuth
from common import list_session_with_chat_assistants
auth = RAGFlowHttpApiAuth("valid_api_token")
chat_assistant_id = "assistant_123"
params = {
"page": 1,
"page_size": 5,
"orderby": "create_time",
"desc": True,
"name": "support_session"
}
response = list_session_with_chat_assistants(auth, chat_assistant_id, params=params)
if response["code"] == 0:
for session in response["data"]:
print(session["id"], session["name"], session["create_time"])
else:
print(f"Error: {response['message']}")
Mermaid Diagram: Class Structure and Test Organization
classDiagram
class TestAuthorization {
+test_invalid_auth(invalid_auth, expected_code, expected_message)
}
class TestSessionsWithChatAssistantList {
+test_page(params, expected_code, expected_page_size, expected_message)
+test_page_size(params, expected_code, expected_page_size, expected_message)
+test_orderby(params, expected_code, assertions, expected_message)
+test_desc(params, expected_code, assertions, expected_message)
+test_name(params, expected_code, expected_num, expected_message)
+test_id(session_id, expected_code, expected_num, expected_message)
+test_name_and_id(session_id, name, expected_code, expected_num, expected_message)
+test_concurrent_list()
+test_invalid_params()
+test_list_chats_after_deleting_associated_chat_assistant()
}
TestSessionsWithChatAssistantList ..> "common.list_session_with_chat_assistants"
TestAuthorization ..> "common.list_session_with_chat_assistants"
TestSessionsWithChatAssistantList ..> "common.delete_chat_assistants"
TestAuthorization ..> "libs.auth.RAGFlowHttpApiAuth"
TestSessionsWithChatAssistantList ..> "libs.auth.RAGFlowHttpApiAuth"
Summary
This test suite is critical in maintaining the reliability and correctness of the session listing features associated with chat assistants. It ensures that:
Authorization is enforced correctly.
Pagination, filtering, and sorting parameters behave as expected.
The API gracefully handles invalid inputs.
Concurrency does not lead to inconsistent results or crashes.
After deletion of chat assistants, appropriate access restrictions are enforced.
The usage of pytest parametric tests and concurrency testing indicates a thorough approach to quality assurance for this part of the system.