File size: 2,592 Bytes
20f348c |
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 |
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)
|