added teardown step of clearing test bucke to storage tests; added tests for nested paths

This commit is contained in:
Matthias Bisping 2022-02-25 11:31:15 +01:00
parent 6e26c1a696
commit 23b35b1f91

View File

@ -14,7 +14,10 @@ from pyinfra.test.storage.adapter_mock import StorageAdapterMock
from pyinfra.test.storage.client_mock import StorageClientMock
# TODO Add test for recursive clearing & getting
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"])
class TestStorage:
def test_clearing_bucket_yields_empty_bucket(self, storage, bucket_name):
@ -27,9 +30,14 @@ class TestStorage:
data_received = storage.get(bucket_name, "file")
assert b"content" == data_received
def test_getting_nested_object_put_in_bucket_is_nested_object(self, storage, bucket_name):
storage.put(bucket_name, "folder/file", b"content")
data_received = storage.get(bucket_name, "folder/file")
assert b"content" == data_received
def test_getting_objects_put_in_bucket_are_objects(self, storage, bucket_name):
storage.put(bucket_name, "file1", b"content 1")
storage.put(bucket_name, "file2", b"content 2")
storage.put(bucket_name, "folder/file2", b"content 2")
data_received = storage.get_all(bucket_name)
assert {b"content 1", b"content 2"} == {*data_received}
@ -40,8 +48,8 @@ class TestStorage:
def test_listing_bucket_files_yields_all_files_in_bucket(self, storage, bucket_name):
storage.put(bucket_name, "file1", b"content 1")
storage.put(bucket_name, "file2", b"content 2")
full_names_recived = storage.list_bucket_files(bucket_name)
assert {(bucket_name, "file1"), (bucket_name, "file2")} == {*full_names_recived}
full_names_received = storage.list_bucket_files(bucket_name)
assert {(bucket_name, "file1"), (bucket_name, "file2")} == {*full_names_received}
def get_adapter(client_name, s3_backend):
@ -57,11 +65,13 @@ def get_adapter(client_name, s3_backend):
@pytest.fixture(params=["minio", "aws"])
def storage(client_name, bucket_name, request):
logger.debug("Setup for storage")
storage = Storage(get_adapter(client_name, request.param))
storage.make_bucket(bucket_name)
storage.clear_bucket(bucket_name)
logging.info(client_name)
return storage
yield storage
logger.debug("Teardown for storage")
storage.clear_bucket(bucket_name)
def test_get_azure_storage_yields_storage():