mirror of
https://github.com/langgenius/dify.git
synced 2025-11-04 04:43:09 +00:00
63 lines
1.5 KiB
Python
63 lines
1.5 KiB
Python
from collections.abc import Generator
|
|
|
|
import pytest
|
|
|
|
from extensions.storage.base_storage import BaseStorage
|
|
|
|
|
|
def get_example_folder() -> str:
|
|
return "~/dify"
|
|
|
|
|
|
def get_example_bucket() -> str:
|
|
return "dify"
|
|
|
|
|
|
def get_opendal_bucket() -> str:
|
|
return "./dify"
|
|
|
|
|
|
def get_example_filename() -> str:
|
|
return "test.txt"
|
|
|
|
|
|
def get_example_data() -> bytes:
|
|
return b"test"
|
|
|
|
|
|
def get_example_filepath() -> str:
|
|
return "~/test"
|
|
|
|
|
|
class BaseStorageTest:
|
|
@pytest.fixture(autouse=True)
|
|
def setup_method(self, *args, **kwargs):
|
|
"""Should be implemented in child classes to setup specific storage."""
|
|
self.storage: BaseStorage
|
|
|
|
def test_save(self):
|
|
"""Test saving data."""
|
|
self.storage.save(get_example_filename(), get_example_data())
|
|
|
|
def test_load_once(self):
|
|
"""Test loading data once."""
|
|
assert self.storage.load_once(get_example_filename()) == get_example_data()
|
|
|
|
def test_load_stream(self):
|
|
"""Test loading data as a stream."""
|
|
generator = self.storage.load_stream(get_example_filename())
|
|
assert isinstance(generator, Generator)
|
|
assert next(generator) == get_example_data()
|
|
|
|
def test_download(self):
|
|
"""Test downloading data."""
|
|
self.storage.download(get_example_filename(), get_example_filepath())
|
|
|
|
def test_exists(self):
|
|
"""Test checking if a file exists."""
|
|
assert self.storage.exists(get_example_filename())
|
|
|
|
def test_delete(self):
|
|
"""Test deleting a file."""
|
|
self.storage.delete(get_example_filename())
|