sdk/python/test/test_document.py

from api.settings import RetCode
from test_sdkbase import TestSdk
from ragflow import RAGFlow
import pytest
from common import API_KEY, HOST_ADDRESS


class TestFile(TestSdk):
    """
    This class contains a suite of tests for the content management functionality within the dataset.
    It ensures that the following functionalities as expected:
        1. upload local files
        2. upload remote files
        3. download a file
        4. delete a file
        5. enable rename
        6. list files
        7. start parsing
        8. end parsing
        9. check the status of the file
        10. list the chunks
        11. delete a chunk
        12. insert a new chunk
        13. edit the status of chunk
        14. get the specific chunk
        15. retrieval test
    """

# ----------------------------upload local files-----------------------------------------------------
    def test_upload_two_files(self):
        """
        Test uploading two files with success.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_two_files")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.SUCCESS and res["message"] == "success"

    def test_upload_one_file(self):
        """
        Test uploading one file with success.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_one_file")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.SUCCESS and res["message"] == "success"

    def test_upload_nonexistent_files(self):
        """
        Test uploading a file which does not exist.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_nonexistent_files")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/imagination.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.DATA_ERROR and "does not exist" in res["message"]

    def test_upload_file_if_dataset_does_not_exist(self):
        """
        Test uploading files if the dataset id does not exist.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        file_paths = ["test_data/test.txt"]
        res = ragflow.upload_local_file("111", file_paths)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "Can't find this dataset"

    def test_upload_file_without_name(self):
        """
        Test uploading files that do not have name.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_file_without_name")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.SUCCESS

    def test_upload_file_without_name1(self):
        """
        Test uploading files that do not have name.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_file_without_name")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/.txt", "test_data/empty.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.SUCCESS

    def test_upload_files_exceeding_the_number_limit(self):
        """
        Test uploading files whose number exceeds the limit.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_files_exceeding_the_number_limit")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/test1.txt"] * 256
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert (res["message"] ==
                "You try to upload 512 files, which exceeds the maximum number of uploading files: 256"
                and res["code"] == RetCode.DATA_ERROR)

    def test_upload_files_without_files(self):
        """
        Test uploading files without files.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_files_without_files")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = [None]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert (res["message"] == "None is not string." and res["code"] == RetCode.ARGUMENT_ERROR)

    def test_upload_files_with_two_files_with_same_name(self):
        """
        Test uploading files with the same name.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_files_with_two_files_with_same_name")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"] * 2
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert (res["message"] == "success" and res["code"] == RetCode.SUCCESS)

    def test_upload_files_with_file_paths(self):
        """
        Test uploading files with only specifying the file path's repo.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_files_with_file_paths")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert (res["message"] == "The file test_data/ does not exist" and res["code"] == RetCode.DATA_ERROR)

    def test_upload_files_with_remote_file_path(self):
        """
        Test uploading files with remote files.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_upload_files_with_remote_file_path")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["https://github.com/genostack/ragflow"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        assert res["code"] == RetCode.ARGUMENT_ERROR and res["message"] == "Remote files have not unsupported."

# ----------------------------delete a file-----------------------------------------------------
    def test_delete_one_file(self):
        """
        Test deleting one file with success.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_delete_one_file")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        # get the doc_id
        data = res["data"][0]
        doc_id = data["id"]
        # delete the files
        deleted_res = ragflow.delete_files(doc_id, dataset_id)
        # assert value
        assert deleted_res["code"] == RetCode.SUCCESS and deleted_res["data"] is True

    def test_delete_document_with_not_existing_document(self):
        """
        Test deleting a document that does not exist with failure.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_delete_document_with_not_existing_document")
        dataset_id = created_res["data"]["dataset_id"]
        res = ragflow.delete_files("111", dataset_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "Document 111 not found!"

    def test_delete_document_with_creating_100_documents_and_deleting_100_documents(self):
        """
        Test deleting documents when uploading 100 docs and deleting 100 docs.
        """
        # upload 100 docs
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_delete_one_file")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"] * 100
        res = ragflow.upload_local_file(dataset_id, file_paths)

        # get the doc_id
        data = res["data"]
        for d in data:
            doc_id = d["id"]
            # delete the files
            deleted_res = ragflow.delete_files(doc_id, dataset_id)
            # assert value
            assert deleted_res["code"] == RetCode.SUCCESS and deleted_res["data"] is True

    def test_delete_document_from_nonexistent_dataset(self):
        """
        Test deleting documents from a non-existent dataset
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_delete_one_file")
        dataset_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"]
        res = ragflow.upload_local_file(dataset_id, file_paths)
        # get the doc_id
        data = res["data"][0]
        doc_id = data["id"]
        # delete the files
        deleted_res = ragflow.delete_files(doc_id, "000")
        # assert value
        assert (deleted_res["code"] == RetCode.ARGUMENT_ERROR and deleted_res["message"] ==
                f"The document {doc_id} is not in the dataset: 000, but in the dataset: {dataset_id}.")

    def test_delete_document_which_is_located_in_other_dataset(self):
        """
        Test deleting a document which is located in other dataset.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        # upload a document
        created_res = ragflow.create_dataset("test_delete_document_which_is_located_in_other_dataset")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"]
        res = ragflow.upload_local_file(created_res_id, file_paths)
        # other dataset
        other_res = ragflow.create_dataset("other_dataset")
        other_dataset_id = other_res["data"]["dataset_id"]
        # get the doc_id
        data = res["data"][0]
        doc_id = data["id"]
        # delete the files from the other dataset
        deleted_res = ragflow.delete_files(doc_id, other_dataset_id)
        # assert value
        assert (deleted_res["code"] == RetCode.ARGUMENT_ERROR and deleted_res["message"] ==
                f"The document {doc_id} is not in the dataset: {other_dataset_id}, but in the dataset: {created_res_id}.")

# ----------------------------list files-----------------------------------------------------
    def test_list_documents_with_success(self):
        """
        Test listing documents with a successful outcome.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        # upload a document
        created_res = ragflow.create_dataset("test_list_documents_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 1

    def test_list_documents_with_checking_size(self):
        """
        Test listing documents and verify the size and names of the documents.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        # upload 10 documents
        created_res = ragflow.create_dataset("test_list_documents_with_checking_size")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"] * 10
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 10

    def test_list_documents_with_getting_empty_result(self):
        """
        Test listing documents that should be empty.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        # upload 0 documents
        created_res = ragflow.create_dataset("test_list_documents_with_getting_empty_result")
        created_res_id = created_res["data"]["dataset_id"]
        # Call the list_document method
        response = ragflow.list_files(created_res_id)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 0

    def test_list_documents_with_creating_100_documents(self):
        """
        Test listing 100 documents and verify the size of these documents.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        # upload 100 documents
        created_res = ragflow.create_dataset("test_list_documents_with_creating_100_documents")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt"] * 100
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 100

    def test_list_document_with_failure(self):
        """
        Test listing documents with IndexError.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_list_document_with_failure")
        created_res_id = created_res["data"]["dataset_id"]
        response = ragflow.list_files(created_res_id, offset=-1, count=-1)
        assert "IndexError" in response["message"] and response["code"] == RetCode.EXCEPTION_ERROR

    def test_list_document_with_verifying_offset_and_count(self):
        """
        Test listing documents with verifying the functionalities of offset and count.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_list_document_with_verifying_offset_and_count")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/empty.txt"] * 10
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id, offset=2, count=10)

        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 10

    def test_list_document_with_verifying_keywords(self):
        """
        Test listing documents with verifying the functionality of searching keywords.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_list_document_with_verifying_keywords")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/empty.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id, keywords="empty")

        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 1

    def test_list_document_with_verifying_order_by_and_descend(self):
        """
        Test listing documents with verifying the functionality of order_by and descend.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_list_document_with_verifying_order_by_and_descend")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/empty.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 2
        docs = response["data"]["docs"]
        # reverse
        i = 1
        for doc in docs:
            assert doc["name"] in file_paths[i]
            i -= 1

    def test_list_document_with_verifying_order_by_and_ascend(self):
        """
        Test listing documents with verifying the functionality of order_by and ascend.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_list_document_with_verifying_order_by_and_ascend")
        created_res_id = created_res["data"]["dataset_id"]
        file_paths = ["test_data/test.txt", "test_data/test1.txt", "test_data/empty.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        # Call the list_document method
        response = ragflow.list_files(created_res_id, descend=False)
        assert response["code"] == RetCode.SUCCESS and len(response["data"]["docs"]) == 3

        docs = response["data"]["docs"]

        i = 0
        for doc in docs:
            assert doc["name"] in file_paths[i]
            i += 1

# ----------------------------update files: enable, rename, template_type-------------------------------------------

    def test_update_nonexistent_document(self):
        """
        Test updating a document which does not exist.
        """
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        params = {
            "name": "new_name"
        }
        res = ragflow.update_file(created_res_id, "weird_doc_id", **params)
        assert res["code"] == RetCode.ARGUMENT_ERROR and res["message"] == f"This document weird_doc_id cannot be found!"

    def test_update_document_without_parameters(self):
        """
        Test updating a document without giving parameters.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_without_parameters")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.DATA_ERROR and
                update_res["message"] == "Please input at least one parameter that you want to update!")

    def test_update_document_in_nonexistent_dataset(self):
        """
        Test updating a document in the nonexistent dataset.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_in_nonexistent_dataset")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "name": "new_name"
        }
        update_res = ragflow.update_file("fake_dataset_id", doc_id, **params)
        assert (update_res["code"] == RetCode.DATA_ERROR and
                update_res["message"] == f"This dataset fake_dataset_id cannot be found!")

    def test_update_document_with_different_extension_name(self):
        """
        Test the updating of a document with an extension name that differs from its original.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_different_extension_name")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "name": "new_name.doc"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.ARGUMENT_ERROR and
                update_res["message"] == "The extension of file cannot be changed")

    def test_update_document_with_duplicate_name(self):
        """
        Test the updating of a document with a duplicate name.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_different_extension_name")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "name": "test.txt"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.ARGUMENT_ERROR and
                update_res["message"] == "Duplicated document name in the same dataset.")

    def test_update_document_with_updating_its_name_with_success(self):
        """
        Test the updating of a document's name with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_name_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "name": "new_name.txt"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.SUCCESS and
                update_res["message"] == "Success" and update_res["data"]["name"] == "new_name.txt")

    def test_update_document_with_updating_its_template_type_with_success(self):
        """
        Test the updating of a document's template type with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_template_type_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "template_type": "laws"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.SUCCESS and
                update_res["message"] == "Success" and update_res["data"]["parser_id"] == "laws")

    def test_update_document_with_updating_its_enable_value_with_success(self):
        """
        Test the updating of a document's enable value with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_enable_value_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "enable": "0"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.SUCCESS and
                update_res["message"] == "Success" and update_res["data"]["status"] == "0")

    def test_update_document_with_updating_illegal_parameter(self):
        """
        Test the updating of a document's illegal parameter.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_illegal_parameter")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "illegal_parameter": "0"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.ARGUMENT_ERROR and
                update_res["message"] == "illegal_parameter is an illegal parameter.")

    def test_update_document_with_giving_its_name_value(self):
        """
        Test the updating of a document's name without its name value.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_name_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "name": ""
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.DATA_ERROR and
                update_res["message"] == "There is no new name.")

    def test_update_document_with_giving_illegal_value_for_enable(self):
        """
        Test the updating of a document's with giving illegal enable's value.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_name_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "enable": "?"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.DATA_ERROR and
                update_res["message"] == "Illegal value ? for 'enable' field.")

    def test_update_document_with_giving_illegal_value_for_type(self):
        """
        Test the updating of a document's with giving illegal type's value.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_update_document_with_updating_its_name_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # update file
        params = {
            "template_type": "?"
        }
        update_res = ragflow.update_file(created_res_id, doc_id, **params)
        assert (update_res["code"] == RetCode.DATA_ERROR and
                update_res["message"] == "Illegal value ? for 'template_type' field.")

# ----------------------------download a file-----------------------------------------------------

    def test_download_nonexistent_document(self):
        """
        Test downloading a document which does not exist.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        res = ragflow.download_file(created_res_id, "imagination")
        assert res["code"] == RetCode.ARGUMENT_ERROR and res["message"] == f"This document 'imagination' cannot be found!"

    def test_download_document_in_nonexistent_dataset(self):
        """
        Test downloading a document whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # download file
        res = ragflow.download_file("imagination", doc_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == f"This dataset 'imagination' cannot be found!"

    def test_download_document_with_success(self):
        """
        Test the downloading of a document with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # download file
        with open("test_data/test.txt", "rb") as file:
            binary_data = file.read()
        res = ragflow.download_file(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["data"] == binary_data

    def test_download_an_empty_document(self):
        """
        Test the downloading of an empty document.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/empty.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # download file
        res = ragflow.download_file(created_res_id, doc_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This file is empty."

# ----------------------------start parsing-----------------------------------------------------
    def test_start_parsing_document_with_success(self):
        """
        Test the parsing of a document with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_document_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/lol.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse file
        res = ragflow.start_parsing_document(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_parsing_nonexistent_document(self):
        """
        Test the parsing a document which does not exist.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        res = ragflow.start_parsing_document(created_res_id, "imagination")
        assert res["code"] == RetCode.ARGUMENT_ERROR and res["message"] == "This document 'imagination' cannot be found!"

    def test_start_parsing_document_in_nonexistent_dataset(self):
        """
        Test the parsing a document whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse
        res = ragflow.start_parsing_document("imagination", doc_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This dataset 'imagination' cannot be found!"

    def test_start_parsing_an_empty_document(self):
        """
        Test the parsing of an empty document.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/empty.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        res = ragflow.start_parsing_document(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == "Empty data in the document: empty.txt; "

    # ------------------------parsing multiple documents----------------------------
    def test_start_parsing_documents_in_nonexistent_dataset(self):
        """
        Test the parsing documents whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # parse
        res = ragflow.start_parsing_documents("imagination")
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This dataset 'imagination' cannot be found!"

    def test_start_parsing_multiple_documents(self):
        """
        Test the parsing documents with a success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        res = ragflow.start_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["data"] is True and res["message"] == ""

    def test_start_parsing_multiple_documents_with_one_empty_file(self):
        """
        Test the parsing documents, one of which is empty.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt", "test_data/empty.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        res = ragflow.start_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == "Empty data in the document: empty.txt; "

    def test_start_parsing_multiple_specific_documents(self):
        """
        Test the parsing documents whose document ids are specified.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_re_parsing_multiple_specific_documents(self):
        """
        Test the re-parsing documents.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        # re-parse
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_re_parsing_multiple_specific_documents_with_changing_parser_id(self):
        """
        Test the re-parsing documents after changing the parser id.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        # general -> laws
        params = {
            "template_type": "laws"
        }
        ragflow.update_file(created_res_id, doc_ids[0], **params)
        # re-parse
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_re_parsing_multiple_specific_documents_with_changing_illegal_parser_id(self):
        """
        Test the re-parsing documents after changing an illegal parser id.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        # general -> illegal
        params = {
            "template_type": "illegal"
        }
        res = ragflow.update_file(created_res_id, doc_ids[0], **params)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "Illegal value illegal for 'template_type' field."
        # re-parse
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_parsing_multiple_specific_documents_with_changing_illegal_parser_id(self):
        """
        Test the parsing documents after changing an illegal parser id.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        # general -> illegal
        params = {
            "template_type": "illegal"
        }
        res = ragflow.update_file(created_res_id, doc_ids[0], **params)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "Illegal value illegal for 'template_type' field."
        # re-parse
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_start_parsing_multiple_documents_in_the_dataset_whose_parser_id_is_illegal(self):
        """
        Test the parsing documents whose dataset's parser id is illegal.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_multiple_documents_in_the_dataset_whose_parser_id_is_illegal")
        created_res_id = created_res["data"]["dataset_id"]
        # update the parser id
        params = {
            "chunk_method": "illegal"
        }
        res = ragflow.update_dataset("test_start_parsing_multiple_documents_in_the_dataset_whose_parser_id_is_illegal", **params)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "Illegal value illegal for 'chunk_method' field."
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        # parse
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

# ----------------------------stop parsing-----------------------------------------------------
    def test_stop_parsing_document_with_success(self):
        """
        Test the stopping parsing of a document with success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_document_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/lol.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse file
        res = ragflow.start_parsing_document(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        res = ragflow.stop_parsing_document(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""

    def test_stop_parsing_nonexistent_document(self):
        """
        Test the stopping parsing a document which does not exist.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        res = ragflow.stop_parsing_document(created_res_id, "imagination.txt")
        assert res["code"] == RetCode.ARGUMENT_ERROR and res["message"] == "This document 'imagination.txt' cannot be found!"

    def test_stop_parsing_document_in_nonexistent_dataset(self):
        """
        Test the stopping parsing a document whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse
        res = ragflow.stop_parsing_document("imagination", doc_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This dataset 'imagination' cannot be found!"

    # ------------------------stop parsing multiple documents----------------------------
    def test_stop_parsing_documents_in_nonexistent_dataset(self):
        """
        Test the stopping parsing documents whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_download_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # parse
        res = ragflow.stop_parsing_documents("imagination")
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This dataset 'imagination' cannot be found!"

    def test_stop_parsing_multiple_documents(self):
        """
        Test the stopping parsing documents with a success.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        res = ragflow.start_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["data"] is True and res["message"] == ""

        res = ragflow.stop_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["data"] is True and res["message"] == ""

    def test_stop_parsing_multiple_documents_with_one_empty_file(self):
        """
        Test the stopping parsing documents, one of which is empty.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt", "test_data/empty.txt"]
        ragflow.upload_local_file(created_res_id, file_paths)
        res = ragflow.start_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == "Empty data in the document: empty.txt; "
        res = ragflow.stop_parsing_documents(created_res_id)
        assert res["code"] == RetCode.SUCCESS and res["data"] is True and res["message"] == ""

    def test_stop_parsing_multiple_specific_documents(self):
        """
        Test the stopping parsing documents whose document ids are specified.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset(" test_start_parsing_multiple_documents")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt", "test_data/test1.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"]
        doc_ids = []
        for d in data:
            doc_ids.append(d["id"])
        res = ragflow.start_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        res = ragflow.stop_parsing_documents(created_res_id, doc_ids)
        assert res["code"] == RetCode.SUCCESS and res["data"] is True and res["message"] == ""

# ----------------------------show the status of the file-----------------------------------------------------
    def test_show_status_with_success(self):
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_show_status_with_success")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/lol.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse file
        res = ragflow.start_parsing_document(created_res_id, doc_id)
        assert res["code"] == RetCode.SUCCESS and res["message"] == ""
        # show status
        status_res = ragflow.show_parsing_status(created_res_id, doc_id)
        assert status_res["code"] == RetCode.SUCCESS and status_res["data"]["status"] == "RUNNING"

    def test_show_status_nonexistent_document(self):
        """
        Test showing the status of a document which does not exist.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_show_status_nonexistent_document")
        created_res_id = created_res["data"]["dataset_id"]
        res = ragflow.show_parsing_status(created_res_id, "imagination")
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This document: 'imagination' is not a valid document."

    def test_show_status_document_in_nonexistent_dataset(self):
        """
        Test showing the status of a document whose dataset is nonexistent.
        """
        # create a dataset
        ragflow = RAGFlow(API_KEY, HOST_ADDRESS)
        created_res = ragflow.create_dataset("test_show_status_document_in_nonexistent_dataset")
        created_res_id = created_res["data"]["dataset_id"]
        # upload files
        file_paths = ["test_data/test.txt"]
        uploading_res = ragflow.upload_local_file(created_res_id, file_paths)
        # get the doc_id
        data = uploading_res["data"][0]
        doc_id = data["id"]
        # parse
        res = ragflow.show_parsing_status("imagination", doc_id)
        assert res["code"] == RetCode.DATA_ERROR and res["message"] == "This dataset: 'imagination' cannot be found!"
# ----------------------------list the chunks of the file-----------------------------------------------------

# ----------------------------delete the chunk-----------------------------------------------------

# ----------------------------edit the status of the chunk-----------------------------------------------------

# ----------------------------insert a new chunk-----------------------------------------------------

# ----------------------------get a specific chunk-----------------------------------------------------

# ----------------------------retrieval test-----------------------------------------------------