|
from test_sdkbase import TestSdk |
|
import ragflow |
|
from ragflow.ragflow import RAGFLow |
|
import pytest |
|
from unittest.mock import MagicMock |
|
|
|
|
|
class TestCase(TestSdk): |
|
|
|
@pytest.fixture |
|
def ragflow_instance(self): |
|
|
|
return ragflow.ragflow.RAGFLow('123', 'url') |
|
|
|
def test_version(self): |
|
print(ragflow.__version__) |
|
|
|
def test_create_dataset(self): |
|
assert ragflow.ragflow.RAGFLow('123', 'url').create_dataset('abc') == 'abc' |
|
|
|
def test_delete_dataset(self): |
|
assert ragflow.ragflow.RAGFLow('123', 'url').delete_dataset('abc') == 'abc' |
|
|
|
def test_list_dataset_success(self, ragflow_instance, monkeypatch): |
|
|
|
mock_response = MagicMock() |
|
mock_response.status_code = 200 |
|
mock_response.json.return_value = {'datasets': [{'id': 1, 'name': 'dataset1'}, {'id': 2, 'name': 'dataset2'}]} |
|
|
|
|
|
monkeypatch.setattr("requests.get", MagicMock(return_value=mock_response)) |
|
|
|
|
|
result = ragflow_instance.list_dataset() |
|
|
|
|
|
assert result == [{'id': 1, 'name': 'dataset1'}, {'id': 2, 'name': 'dataset2'}] |
|
|
|
def test_list_dataset_failure(self, ragflow_instance, monkeypatch): |
|
|
|
mock_response = MagicMock() |
|
mock_response.status_code = 404 |
|
|
|
|
|
monkeypatch.setattr("requests.get", MagicMock(return_value=mock_response)) |
|
|
|
|
|
result = ragflow_instance.list_dataset() |
|
|
|
|
|
assert result is None |
|
|