Test: Added test cases for Create Session With Chat Assistant HTTP API (#6902)

### What problem does this PR solve?

cover [create session with chat
assistant](https://ragflow.io/docs/dev/http_api_reference#create-session-with-chat-assistant)
endpoints

### Type of change

- [x] add test cases

sdk/python/test/test_http_api/common.py

 FILE_CHUNK_API_URL = "/api/v1/datasets/{dataset_id}/chunks"
 CHUNK_API_URL = "/api/v1/datasets/{dataset_id}/documents/{document_id}/chunks"
 CHAT_ASSISTANT_API_URL = "/api/v1/chats"
+SESSION_WITH_CHAT_ASSISTANT_API_URL = "/api/v1/chats/{chat_id}/sessions"
+SESSION_WITH_AGENT_API_URL = "/api/v1/agents/{agent_id}/sessions"
 
 INVALID_API_TOKEN = "invalid_key_123"
 DATASET_NAME_LIMIT = 128
 DOCUMENT_NAME_LIMIT = 128
-CHAT_ASSISTANT_LIMIT = 255
+CHAT_ASSISTANT_NAME_LIMIT = 255
+SESSION_WITH_CHAT_NAME_LIMIT = 255
 
 
 # DATASET MANAGEMENT
         res = create_chat_assistant(auth, {"name": f"test_chat_assistant_{i}", "dataset_ids": []})
         chat_assistant_ids.append(res["data"]["id"])
     return chat_assistant_ids
+
+
+# SESSION MANAGEMENT
+def create_session_with_chat_assistant(auth, chat_assistant_id, payload=None):
+    url = f"{HOST_ADDRESS}{SESSION_WITH_CHAT_ASSISTANT_API_URL}".format(chat_id=chat_assistant_id)
+    res = requests.post(url=url, headers=HEADERS, auth=auth, json=payload)
+    return res.json()
+
+
+def list_session_with_chat_assistants(auth, chat_assistant_id, params=None):
+    url = f"{HOST_ADDRESS}{SESSION_WITH_CHAT_ASSISTANT_API_URL}".format(chat_id=chat_assistant_id)
+    res = requests.get(url=url, headers=HEADERS, auth=auth, params=params)
+    return res.json()
+
+
+def update_session_with_chat_assistant(auth, chat_assistant_id, session_id, payload=None):
+    url = f"{HOST_ADDRESS}{SESSION_WITH_CHAT_ASSISTANT_API_URL}/{session_id}".format(chat_id=chat_assistant_id)
+    res = requests.put(url=url, headers=HEADERS, auth=auth, json=payload)
+    return res.json()
+
+
+def delete_session_with_chat_assistants(auth, chat_assistant_id, payload=None):
+    url = f"{HOST_ADDRESS}{SESSION_WITH_CHAT_ASSISTANT_API_URL}".format(chat_id=chat_assistant_id)
+    res = requests.delete(url=url, headers=HEADERS, auth=auth, json=payload)
+    return res.json()

sdk/python/test/test_http_api/conftest.py

 #
 
 import pytest
-from common import add_chunk, batch_create_datasets, bulk_upload_documents, delete_chat_assistants, delete_datasets, list_documnets, parse_documnets
+from common import (
+    add_chunk,
+    batch_create_datasets,
+    bulk_upload_documents,
+    create_chat_assistant,
+    delete_chat_assistants,
+    delete_datasets,
+    delete_session_with_chat_assistants,
+    list_documnets,
+    parse_documnets,
+)
 from libs.utils import wait_for
 from libs.utils.file_utils import (
     create_docx_file,
 
 
 @pytest.fixture(scope="function")
-def clear_datasets(get_http_api_auth):
-    yield
-    delete_datasets(get_http_api_auth)
+def clear_datasets(request, get_http_api_auth):
+    def cleanup():
+        delete_datasets(get_http_api_auth)
+
+    request.addfinalizer(cleanup)
+
+
+@pytest.fixture(scope="function")
+def clear_chat_assistants(request, get_http_api_auth):
+    def cleanup():
+        delete_chat_assistants(get_http_api_auth)
+
+    request.addfinalizer(cleanup)
 
 
 @pytest.fixture(scope="function")
-def clear_chat_assistants(get_http_api_auth):
-    yield
-    delete_chat_assistants(get_http_api_auth)
+def clear_session_with_chat_assistants(request, get_http_api_auth, add_chat_assistants):
+    _, _, chat_assistant_ids = add_chat_assistants
+
+    def cleanup():
+        for chat_assistant_id in chat_assistant_ids:
+            delete_session_with_chat_assistants(get_http_api_auth, chat_assistant_id)
+
+    request.addfinalizer(cleanup)
 
 
 @pytest.fixture
 
     sleep(1)
     return dataset_id, document_id, chunk_ids
+
+
+@pytest.fixture(scope="class")
+def add_chat_assistants(request, get_http_api_auth, add_document):
+    def cleanup():
+        delete_chat_assistants(get_http_api_auth)
+
+    request.addfinalizer(cleanup)
+
+    dataset_id, document_id = add_document
+    parse_documnets(get_http_api_auth, dataset_id, {"document_ids": [document_id]})
+    condition(get_http_api_auth, dataset_id)
+
+    chat_assistant_ids = []
+    for i in range(5):
+        res = create_chat_assistant(get_http_api_auth, {"name": f"test_chat_assistant_{i}", "dataset_ids": [dataset_id]})
+        chat_assistant_ids.append(res["data"]["id"])
+
+    return dataset_id, document_id, chat_assistant_ids

sdk/python/test/test_http_api/test_chat_assistant_management/conftest.py

 #  limitations under the License.
 #
 import pytest
-from common import create_chat_assistant, delete_chat_assistants
+from common import create_chat_assistant, delete_chat_assistants, list_documnets, parse_documnets
+from libs.utils import wait_for
 
 
-@pytest.fixture(scope="class")
-def add_chat_assistants(request, get_http_api_auth, add_chunks):
-    def cleanup():
-        delete_chat_assistants(get_http_api_auth)
-
-    request.addfinalizer(cleanup)
-
-    dataset_id, document_id, chunk_ids = add_chunks
-    chat_assistant_ids = []
-    for i in range(5):
-        res = create_chat_assistant(get_http_api_auth, {"name": f"test_chat_assistant_{i}", "dataset_ids": [dataset_id]})
-        chat_assistant_ids.append(res["data"]["id"])
-
-    return dataset_id, document_id, chunk_ids, chat_assistant_ids
+@wait_for(30, 1, "Document parsing timeout")
+def condition(_auth, _dataset_id):
+    res = list_documnets(_auth, _dataset_id)
+    for doc in res["data"]["docs"]:
+        if doc["run"] != "DONE":
+            return False
+    return True
 
 
 @pytest.fixture(scope="function")
-def add_chat_assistants_func(request, get_http_api_auth, add_chunks):
+def add_chat_assistants_func(request, get_http_api_auth, add_document):
     def cleanup():
         delete_chat_assistants(get_http_api_auth)
 
     request.addfinalizer(cleanup)
 
-    dataset_id, document_id, chunk_ids = add_chunks
+    dataset_id, document_id = add_document
+    parse_documnets(get_http_api_auth, dataset_id, {"document_ids": [document_id]})
+    condition(get_http_api_auth, dataset_id)
+
     chat_assistant_ids = []
     for i in range(5):
         res = create_chat_assistant(get_http_api_auth, {"name": f"test_chat_assistant_{i}", "dataset_ids": [dataset_id]})
         chat_assistant_ids.append(res["data"]["id"])
 
-    return dataset_id, document_id, chunk_ids, chat_assistant_ids
+    return dataset_id, document_id, chat_assistant_ids

sdk/python/test/test_http_api/test_chat_assistant_management/test_create_chat_assistant.py

 #
 
 import pytest
-from common import CHAT_ASSISTANT_LIMIT, INVALID_API_TOKEN, create_chat_assistant
+from common import CHAT_ASSISTANT_NAME_LIMIT, INVALID_API_TOKEN, create_chat_assistant
 from libs.auth import RAGFlowHttpApiAuth
 from libs.utils import encode_avatar
 from libs.utils.file_utils import create_image_file
         "payload, expected_code, expected_message",
         [
             ({"name": "valid_name"}, 0, ""),
-            pytest.param({"name": "a" * (CHAT_ASSISTANT_LIMIT + 1)}, 102, "", marks=pytest.mark.skip(reason="issues/")),
+            pytest.param({"name": "a" * (CHAT_ASSISTANT_NAME_LIMIT + 1)}, 102, "", marks=pytest.mark.skip(reason="issues/")),
             pytest.param({"name": 1}, 100, "", marks=pytest.mark.skip(reason="issues/")),
             ({"name": ""}, 102, "`name` is required."),
             ({"name": "duplicated_name"}, 102, "Duplicated chat name in creating chat."),

sdk/python/test/test_http_api/test_chat_assistant_management/test_delete_chat_assistants.py

         ],
     )
     def test_basic_scenarios(self, get_http_api_auth, add_chat_assistants_func, payload, expected_code, expected_message, remaining):
-        _, _, _, chat_assistant_ids = add_chat_assistants_func
+        _, _, chat_assistant_ids = add_chat_assistants_func
         if callable(payload):
             payload = payload(chat_assistant_ids)
         res = delete_chat_assistants(get_http_api_auth, payload)
         ],
     )
     def test_delete_partial_invalid_id(self, get_http_api_auth, add_chat_assistants_func, payload):
-        _, _, _, chat_assistant_ids = add_chat_assistants_func
+        _, _, chat_assistant_ids = add_chat_assistants_func
         if callable(payload):
             payload = payload(chat_assistant_ids)
         res = delete_chat_assistants(get_http_api_auth, payload)
         assert len(res["data"]) == 0
 
     def test_repeated_deletion(self, get_http_api_auth, add_chat_assistants_func):
-        _, _, _, chat_assistant_ids = add_chat_assistants_func
+        _, _, chat_assistant_ids = add_chat_assistants_func
         res = delete_chat_assistants(get_http_api_auth, {"ids": chat_assistant_ids})
         assert res["code"] == 0
 
 
     @pytest.mark.skip(reason="issues/6876")
     def test_duplicate_deletion(self, get_http_api_auth, add_chat_assistants_func):
-        _, _, _, chat_assistant_ids = add_chat_assistants_func
+        _, _, chat_assistant_ids = add_chat_assistants_func
         res = delete_chat_assistants(get_http_api_auth, {"ids": chat_assistant_ids + chat_assistant_ids})
         print(res)
         assert res["code"] == 0
         res = delete_chat_assistants(get_http_api_auth)
         assert len(res["data"]) == 0
 
+    @pytest.mark.slow
     def test_concurrent_deletion(self, get_http_api_auth):
         ids = batch_create_chat_assistants(get_http_api_auth, 100)
 

sdk/python/test/test_http_api/test_chat_assistant_management/test_list_chat_assistants.py

         expected_num,
         expected_message,
     ):
-        _, _, _, chat_assistant_ids = add_chat_assistants
+        _, _, chat_assistant_ids = add_chat_assistants
         if callable(chat_assistant_id):
             params = {"id": chat_assistant_id(chat_assistant_ids)}
         else:
         expected_num,
         expected_message,
     ):
-        _, _, _, chat_assistant_ids = add_chat_assistants
+        _, _, chat_assistant_ids = add_chat_assistants
         if callable(chat_assistant_id):
             params = {"id": chat_assistant_id(chat_assistant_ids), "name": name}
         else:
         else:
             assert res["message"] == expected_message
 
+    @pytest.mark.slow
     def test_concurrent_list(self, get_http_api_auth):
         with ThreadPoolExecutor(max_workers=5) as executor:
             futures = [executor.submit(list_chat_assistants, get_http_api_auth) for i in range(100)]
         assert len(res["data"]) == 5
 
     def test_list_chats_after_deleting_associated_dataset(self, get_http_api_auth, add_chat_assistants):
-        dataset_id, _, _, _ = add_chat_assistants
+        dataset_id, _, _ = add_chat_assistants
         res = delete_datasets(get_http_api_auth, {"ids": [dataset_id]})
         assert res["code"] == 0
 

sdk/python/test/test_http_api/test_chat_assistant_management/test_update_chat_assistant.py

 #  limitations under the License.
 #
 import pytest
-from common import CHAT_ASSISTANT_LIMIT, INVALID_API_TOKEN, list_chat_assistants, update_chat_assistant
+from common import CHAT_ASSISTANT_NAME_LIMIT, INVALID_API_TOKEN, list_chat_assistants, update_chat_assistant
 from libs.auth import RAGFlowHttpApiAuth
 from libs.utils import encode_avatar
 from libs.utils.file_utils import create_image_file
         "payload, expected_code, expected_message",
         [
             ({"name": "valid_name"}, 0, ""),
-            pytest.param({"name": "a" * (CHAT_ASSISTANT_LIMIT + 1)}, 102, "", marks=pytest.mark.skip(reason="issues/")),
+            pytest.param({"name": "a" * (CHAT_ASSISTANT_NAME_LIMIT + 1)}, 102, "", marks=pytest.mark.skip(reason="issues/")),
             pytest.param({"name": 1}, 100, "", marks=pytest.mark.skip(reason="issues/")),
             ({"name": ""}, 102, "`name` cannot be empty."),
             ({"name": "test_chat_assistant_1"}, 102, "Duplicated chat name in updating chat."),
         ],
     )
     def test_name(self, get_http_api_auth, add_chat_assistants_func, payload, expected_code, expected_message):
-        _, _, _, chat_assistant_ids = add_chat_assistants_func
+        _, _, chat_assistant_ids = add_chat_assistants_func
 
         res = update_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
         assert res["code"] == expected_code, res
         ],
     )
     def test_dataset_ids(self, get_http_api_auth, add_chat_assistants_func, dataset_ids, expected_code, expected_message):
-        dataset_id, _, _, chat_assistant_ids = add_chat_assistants_func
+        dataset_id, _, chat_assistant_ids = add_chat_assistants_func
         payload = {"name": "ragflow test"}
         if callable(dataset_ids):
             payload["dataset_ids"] = dataset_ids(dataset_id)
             assert res["message"] == expected_message
 
     def test_avatar(self, get_http_api_auth, add_chat_assistants_func, tmp_path):
-        dataset_id, _, _, chat_assistant_ids = add_chat_assistants_func
+        dataset_id, _, chat_assistant_ids = add_chat_assistants_func
         fn = create_image_file(tmp_path / "ragflow_test.png")
         payload = {"name": "avatar_test", "avatar": encode_avatar(fn), "dataset_ids": [dataset_id]}
         res = update_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
         ],
     )
     def test_llm(self, get_http_api_auth, add_chat_assistants_func, llm, expected_code, expected_message):
-        dataset_id, _, _, chat_assistant_ids = add_chat_assistants_func
+        dataset_id, _, chat_assistant_ids = add_chat_assistants_func
         payload = {"name": "llm_test", "dataset_ids": [dataset_id], "llm": llm}
         res = update_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
         assert res["code"] == expected_code
         ],
     )
     def test_prompt(self, get_http_api_auth, add_chat_assistants_func, prompt, expected_code, expected_message):
-        dataset_id, _, _, chat_assistant_ids = add_chat_assistants_func
+        dataset_id, _, chat_assistant_ids = add_chat_assistants_func
         payload = {"name": "prompt_test", "dataset_ids": [dataset_id], "prompt": prompt}
         res = update_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
         assert res["code"] == expected_code

sdk/python/test/test_http_api/test_chunk_management_within_dataset/test_add_chunk.py

         assert res["code"] == 102
         assert res["message"] == f"You don't own the document {document_id}."
 
+    @pytest.mark.slow
     @pytest.mark.skip(reason="issues/6411")
     def test_concurrent_add_chunk(self, get_http_api_auth, add_document):
         chunk_num = 50

sdk/python/test/test_http_api/test_chunk_management_within_dataset/test_retrieval_chunks.py

         assert res["code"] == 0
         assert len(res["data"]["chunks"]) == 4
 
+    @pytest.mark.slow
     def test_concurrent_retrieval(self, get_http_api_auth, add_chunks):
         from concurrent.futures import ThreadPoolExecutor
 

sdk/python/test/test_http_api/test_chunk_management_within_dataset/test_update_chunk.py

         if expected_code != 0:
             assert res["message"] == expected_message
 
+    @pytest.mark.slow
     @pytest.mark.skipif(os.getenv("DOC_ENGINE") == "infinity", reason="issues/6554")
     def test_concurrent_update_chunk(self, get_http_api_auth, add_chunks):
         chunk_num = 50

sdk/python/test/test_http_api/test_dataset_mangement/test_delete_datasets.py

         res = list_datasets(get_http_api_auth)
         assert len(res["data"]) == 0
 
+    @pytest.mark.slow
     def test_concurrent_deletion(self, get_http_api_auth):
         ids = batch_create_datasets(get_http_api_auth, 100)
 

sdk/python/test/test_http_api/test_dataset_mangement/test_list_datasets.py

         else:
             assert res["message"] == expected_message
 
+    @pytest.mark.slow
     def test_concurrent_list(self, get_http_api_auth):
         with ThreadPoolExecutor(max_workers=5) as executor:
             futures = [executor.submit(list_datasets, get_http_api_auth) for i in range(100)]

sdk/python/test/test_http_api/test_dataset_mangement/test_update_dataset.py

         res = update_dataset(get_http_api_auth, dataset_id, {"unknown_field": 0})
         assert res["code"] == 100
 
+    @pytest.mark.slow
     def test_concurrent_update(self, get_http_api_auth, add_dataset_func):
         dataset_id = add_dataset_func
 

sdk/python/test/test_http_api/test_file_management_within_dataset/test_delete_documents.py

         assert res["data"]["total"] == 0
 
 
+@pytest.mark.slow
 def test_concurrent_deletion(get_http_api_auth, add_dataset, tmp_path):
     documnets_num = 100
     dataset_id = add_dataset

sdk/python/test/test_http_api/test_file_management_within_dataset/test_download_document.py

             )
 
 
+@pytest.mark.slow
 def test_concurrent_download(get_http_api_auth, add_dataset, tmp_path):
     document_count = 20
     dataset_id = add_dataset

sdk/python/test/test_http_api/test_file_management_within_dataset/test_list_documents.py

         else:
             assert res["message"] == expected_message
 
+    @pytest.mark.slow
     def test_concurrent_list(self, get_http_api_auth, add_documents):
         dataset_id, _ = add_documents
 

sdk/python/test/test_http_api/test_file_management_within_dataset/test_upload_documents.py

         res = list_datasets(get_http_api_auth, {"id": dataset_id})
         assert res["data"][0]["document_count"] == expected_document_count
 
+    @pytest.mark.slow
     def test_concurrent_upload(self, get_http_api_auth, add_dataset_func, tmp_path):
         dataset_id = add_dataset_func
 

sdk/python/test/test_http_api/test_session_management/test_create_session_with_chat_assistant.py

+#
+#  Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from common import INVALID_API_TOKEN, SESSION_WITH_CHAT_NAME_LIMIT, create_session_with_chat_assistant, delete_chat_assistants, list_session_with_chat_assistants
+from libs.auth import RAGFlowHttpApiAuth
+
+
+class TestAuthorization:
+    @pytest.mark.parametrize(
+        "auth, expected_code, expected_message",
+        [
+            (None, 0, "`Authorization` can't be empty"),
+            (
+                RAGFlowHttpApiAuth(INVALID_API_TOKEN),
+                109,
+                "Authentication error: API key is invalid!",
+            ),
+        ],
+    )
+    def test_invalid_auth(self, auth, expected_code, expected_message):
+        res = create_session_with_chat_assistant(auth, "chat_assistant_id")
+        assert res["code"] == expected_code
+        assert res["message"] == expected_message
+
+
+@pytest.mark.usefixtures("clear_session_with_chat_assistants")
+class TestSessionWithChatAssistantCreate:
+    @pytest.mark.parametrize(
+        "payload, expected_code, expected_message",
+        [
+            ({"name": "valid_name"}, 0, ""),
+            pytest.param({"name": "a" * (SESSION_WITH_CHAT_NAME_LIMIT + 1)}, 102, "", marks=pytest.mark.skip(reason="issues/")),
+            pytest.param({"name": 1}, 100, "", marks=pytest.mark.skip(reason="issues/")),
+            ({"name": ""}, 102, "`name` can not be empty."),
+            ({"name": "duplicated_name"}, 0, ""),
+            ({"name": "case insensitive"}, 0, ""),
+        ],
+    )
+    def test_name(self, get_http_api_auth, add_chat_assistants, payload, expected_code, expected_message):
+        _, _, chat_assistant_ids = add_chat_assistants
+        if payload["name"] == "duplicated_name":
+            create_session_with_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
+        elif payload["name"] == "case insensitive":
+            create_session_with_chat_assistant(get_http_api_auth, chat_assistant_ids[0], {"name": payload["name"].upper()})
+
+        res = create_session_with_chat_assistant(get_http_api_auth, chat_assistant_ids[0], payload)
+        assert res["code"] == expected_code, res
+        if expected_code == 0:
+            assert res["data"]["name"] == payload["name"]
+            assert res["data"]["chat_id"] == chat_assistant_ids[0]
+        else:
+            assert res["message"] == expected_message
+
+    @pytest.mark.parametrize(
+        "chat_assistant_id, expected_code, expected_message",
+        [
+            ("", 100, "<MethodNotAllowed '405: Method Not Allowed'>"),
+            ("invalid_chat_assistant_id", 102, "You do not own the assistant."),
+        ],
+    )
+    def test_invalid_chat_assistant_id(self, get_http_api_auth, chat_assistant_id, expected_code, expected_message):
+        res = create_session_with_chat_assistant(get_http_api_auth, chat_assistant_id, {"name": "valid_name"})
+        assert res["code"] == expected_code
+        assert res["message"] == expected_message
+
+    @pytest.mark.slow
+    def test_concurrent_create_session(self, get_http_api_auth, add_chat_assistants):
+        chunk_num = 1000
+        _, _, chat_assistant_ids = add_chat_assistants
+        res = list_session_with_chat_assistants(get_http_api_auth, chat_assistant_ids[0])
+        if res["code"] != 0:
+            assert False, res
+        chunks_count = len(res["data"])
+
+        with ThreadPoolExecutor(max_workers=5) as executor:
+            futures = [
+                executor.submit(
+                    create_session_with_chat_assistant,
+                    get_http_api_auth,
+                    chat_assistant_ids[0],
+                    {"name": f"session with chat assistant test {i}"},
+                )
+                for i in range(chunk_num)
+            ]
+        responses = [f.result() for f in futures]
+        assert all(r["code"] == 0 for r in responses)
+        res = list_session_with_chat_assistants(get_http_api_auth, chat_assistant_ids[0], {"page_size": chunk_num})
+        if res["code"] != 0:
+            assert False, res
+        assert len(res["data"]) == chunks_count + chunk_num
+
+    def test_add_session_to_deleted_chat_assistant(self, get_http_api_auth, add_chat_assistants):
+        _, _, chat_assistant_ids = add_chat_assistants
+        res = delete_chat_assistants(get_http_api_auth, {"ids": [chat_assistant_ids[0]]})
+        assert res["code"] == 0
+        res = create_session_with_chat_assistant(get_http_api_auth, chat_assistant_ids[0], {"name": "valid_name"})
+        print(res)
+        assert res["code"] == 102
+        assert res["message"] == "You do not own the assistant."