File size: 2,358 Bytes
c2d189b
172caf6
 
 
 
c2d189b
172caf6
 
 
cd7d2b9
172caf6
c2d189b
172caf6
 
 
 
cd7d2b9
172caf6
 
cd7d2b9
c2d189b
678763e
c2d189b
cd7d2b9
c2d189b
 
cd7d2b9
c2d189b
cd7d2b9
 
c2d189b
 
 
 
cd7d2b9
c2d189b
cd7d2b9
 
c2d189b
 
cd7d2b9
c2d189b
 
cd7d2b9
c2d189b
 
cd7d2b9
c2d189b
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from ragflow import RAGFlow,Session

from common import API_KEY, HOST_ADDRESS


class TestSession:
    def test_create_session(self):
        rag = RAGFlow(API_KEY, HOST_ADDRESS)
        kb = rag.create_dataset(name="test_create_session")
        assistant = rag.create_chat(name="test_create_session", knowledgebases=[kb])
        session = assistant.create_session()
        assert isinstance(session,Session), "Failed to create a session."

    def test_create_chat_with_success(self):
        rag = RAGFlow(API_KEY, HOST_ADDRESS)
        kb = rag.create_dataset(name="test_create_chat")
        assistant = rag.create_chat(name="test_create_chat", knowledgebases=[kb])
        session = assistant.create_session()
        question = "What is AI"
        for ans in session.ask(question, stream=True):
            pass
        assert not ans.content.startswith("**ERROR**"), "Please check this error."

    def test_delete_sessions_with_success(self):
        rag = RAGFlow(API_KEY, HOST_ADDRESS)
        kb = rag.create_dataset(name="test_delete_session")
        assistant = rag.create_chat(name="test_delete_session",knowledgebases=[kb])
        session=assistant.create_session()
        res=assistant.delete_sessions(ids=[session.id])
        assert res is None, "Failed to delete the dataset."

    def test_update_session_with_success(self):
        rag=RAGFlow(API_KEY,HOST_ADDRESS)
        kb=rag.create_dataset(name="test_update_session")
        assistant = rag.create_chat(name="test_update_session",knowledgebases=[kb])
        session=assistant.create_session(name="old session")
        res=session.update({"name":"new session"})
        assert res is None,"Failed to update the session"


    def test_list_sessions_with_success(self):
        rag=RAGFlow(API_KEY,HOST_ADDRESS)
        kb=rag.create_dataset(name="test_list_session")
        assistant=rag.create_chat(name="test_list_session",knowledgebases=[kb])
        assistant.create_session("test_1")
        assistant.create_session("test_2")
        sessions=assistant.list_sessions()
        if isinstance(sessions,list):
            for session in sessions:
                assert isinstance(session,Session),"Non-Session elements exist in the list"
        else :
            assert False,"Failed to retrieve the session list."