mirror of
https://git.mirrors.martin98.com/https://github.com/langgenius/dify.git
synced 2025-05-07 17:09:13 +08:00
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
from collections.abc import Generator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from extensions.storage.opendal_storage import OpenDALStorage
|
|
from tests.unit_tests.oss.__mock.base import (
|
|
get_example_data,
|
|
get_example_filename,
|
|
get_opendal_bucket,
|
|
)
|
|
|
|
|
|
class TestOpenDAL:
|
|
@pytest.fixture(autouse=True)
|
|
def setup_method(self, *args, **kwargs):
|
|
"""Executed before each test method."""
|
|
self.storage = OpenDALStorage(
|
|
scheme="fs",
|
|
root=get_opendal_bucket(),
|
|
)
|
|
|
|
@pytest.fixture(scope="class", autouse=True)
|
|
def teardown_class(self, request):
|
|
"""Clean up after all tests in the class."""
|
|
|
|
def cleanup():
|
|
folder = Path(get_opendal_bucket())
|
|
if folder.exists() and folder.is_dir():
|
|
for item in folder.iterdir():
|
|
if item.is_file():
|
|
item.unlink()
|
|
elif item.is_dir():
|
|
item.rmdir()
|
|
folder.rmdir()
|
|
|
|
return cleanup()
|
|
|
|
def test_save_and_exists(self):
|
|
"""Test saving data and checking existence."""
|
|
filename = get_example_filename()
|
|
data = get_example_data()
|
|
|
|
assert not self.storage.exists(filename)
|
|
self.storage.save(filename, data)
|
|
assert self.storage.exists(filename)
|
|
|
|
def test_load_once(self):
|
|
"""Test loading data once."""
|
|
filename = get_example_filename()
|
|
data = get_example_data()
|
|
|
|
self.storage.save(filename, data)
|
|
loaded_data = self.storage.load_once(filename)
|
|
assert loaded_data == data
|
|
|
|
def test_load_stream(self):
|
|
"""Test loading data as a stream."""
|
|
filename = get_example_filename()
|
|
data = get_example_data()
|
|
|
|
self.storage.save(filename, data)
|
|
generator = self.storage.load_stream(filename)
|
|
assert isinstance(generator, Generator)
|
|
assert next(generator) == data
|
|
|
|
def test_download(self):
|
|
"""Test downloading data to a file."""
|
|
filename = get_example_filename()
|
|
filepath = str(Path(get_opendal_bucket()) / filename)
|
|
data = get_example_data()
|
|
|
|
self.storage.save(filename, data)
|
|
self.storage.download(filename, filepath)
|
|
|
|
def test_delete(self):
|
|
"""Test deleting a file."""
|
|
filename = get_example_filename()
|
|
data = get_example_data()
|
|
|
|
self.storage.save(filename, data)
|
|
assert self.storage.exists(filename)
|
|
|
|
self.storage.delete(filename)
|
|
assert not self.storage.exists(filename)
|