File size: 4,527 Bytes
e9078f4 24da205 e9078f4 49c21eb 8faf53b 49c21eb 8faf53b 49c21eb bc9f8af 49c21eb 121b0b5 49c21eb 8faf53b 49c21eb e9078f4 49c21eb 8faf53b 49c21eb 8faf53b 49c21eb e9078f4 d3d83ec |
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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
from ragflow_sdk import RAGFlow,Agent
from common import HOST_ADDRESS
import pytest
def test_create_session_with_success(get_api_key_fixture):
API_KEY = get_api_key_fixture
rag = RAGFlow(API_KEY, HOST_ADDRESS)
kb = rag.create_dataset(name="test_create_session")
displayed_name = "ragflow.txt"
with open("test_data/ragflow.txt", "rb") as file:
blob = file.read()
document = {"displayed_name":displayed_name,"blob":blob}
documents = []
documents.append(document)
docs= kb.upload_documents(documents)
for doc in docs:
doc.add_chunk("This is a test to add chunk")
assistant=rag.create_chat("test_create_session", dataset_ids=[kb.id])
assistant.create_session()
def test_create_conversation_with_success(get_api_key_fixture):
API_KEY = get_api_key_fixture
rag = RAGFlow(API_KEY, HOST_ADDRESS)
kb = rag.create_dataset(name="test_create_conversation")
displayed_name = "ragflow.txt"
with open("test_data/ragflow.txt", "rb") as file:
blob = file.read()
document = {"displayed_name": displayed_name, "blob": blob}
documents = []
documents.append(document)
docs = kb.upload_documents(documents)
for doc in docs:
doc.add_chunk("This is a test to add chunk")
assistant = rag.create_chat("test_create_conversation", dataset_ids=[kb.id])
session = assistant.create_session()
question = "What is AI"
for ans in session.ask(question):
pass
# assert not ans.content.startswith("**ERROR**"), "Please check this error."
def test_delete_sessions_with_success(get_api_key_fixture):
API_KEY = get_api_key_fixture
rag = RAGFlow(API_KEY, HOST_ADDRESS)
kb = rag.create_dataset(name="test_delete_session")
displayed_name = "ragflow.txt"
with open("test_data/ragflow.txt", "rb") as file:
blob = file.read()
document = {"displayed_name":displayed_name,"blob":blob}
documents = []
documents.append(document)
docs= kb.upload_documents(documents)
for doc in docs:
doc.add_chunk("This is a test to add chunk")
assistant=rag.create_chat("test_delete_session", dataset_ids=[kb.id])
session = assistant.create_session()
assistant.delete_sessions(ids=[session.id])
def test_update_session_with_name(get_api_key_fixture):
API_KEY = get_api_key_fixture
rag = RAGFlow(API_KEY, HOST_ADDRESS)
kb = rag.create_dataset(name="test_update_session")
displayed_name = "ragflow.txt"
with open("test_data/ragflow.txt", "rb") as file:
blob = file.read()
document = {"displayed_name": displayed_name, "blob": blob}
documents = []
documents.append(document)
docs = kb.upload_documents(documents)
for doc in docs:
doc.add_chunk("This is a test to add chunk")
assistant = rag.create_chat("test_update_session", dataset_ids=[kb.id])
session = assistant.create_session(name="old session")
session.update({"name": "new session"})
def test_list_sessions_with_success(get_api_key_fixture):
API_KEY = get_api_key_fixture
rag = RAGFlow(API_KEY, HOST_ADDRESS)
kb = rag.create_dataset(name="test_list_session")
displayed_name = "ragflow.txt"
with open("test_data/ragflow.txt", "rb") as file:
blob = file.read()
document = {"displayed_name":displayed_name,"blob":blob}
documents = []
documents.append(document)
docs= kb.upload_documents(documents)
for doc in docs:
doc.add_chunk("This is a test to add chunk")
assistant=rag.create_chat("test_list_session", dataset_ids=[kb.id])
assistant.create_session("test_1")
assistant.create_session("test_2")
assistant.list_sessions()
@pytest.mark.skip(reason="")
def test_create_agent_session_with_success(get_api_key_fixture):
API_KEY = "ragflow-BkOGNhYjIyN2JiODExZWY5MzVhMDI0Mm"
rag = RAGFlow(API_KEY,HOST_ADDRESS)
Agent.create_session("2e45b5209c1011efa3e90242ac120006", rag)
@pytest.mark.skip(reason="")
def test_create_agent_conversation_with_success(get_api_key_fixture):
API_KEY = "ragflow-BkOGNhYjIyN2JiODExZWY5MzVhMDI0Mm"
rag = RAGFlow(API_KEY,HOST_ADDRESS)
session = Agent.create_session("2e45b5209c1011efa3e90242ac120006", rag)
session.ask("What is this job")
@pytest.mark.skip(reason="")
def test_list_agent_sessions_with_success(get_api_key_fixture):
API_KEY = "ragflow-BkOGNhYjIyN2JiODExZWY5MzVhMDI0Mm"
agent_id = "2710f2269b4611ef8fdf0242ac120006"
rag = RAGFlow(API_KEY,HOST_ADDRESS)
Agent.list_sessions(agent_id,rag)
|