50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
import pytest
|
|
|
|
from pyinfra.storage.storage import Storage
|
|
from pyinfra.storage.storages import get_azure_storage
|
|
from pyinfra.test.storage.adapter_mock import StorageAdapterMock
|
|
from pyinfra.test.storage.client_mock import StorageClientMock
|
|
|
|
|
|
class StorageTester:
|
|
def __init__(self, adapter):
|
|
self.bucket_name = "pyinfra-test-bucket"
|
|
self.storage = Storage(adapter=adapter)
|
|
|
|
def test_purging_bucket_yields_empty_bucket(self):
|
|
self.storage.purge(self.bucket_name)
|
|
data_received = self.storage.get_all(self.bucket_name)
|
|
assert not {*data_received}
|
|
|
|
def test_getting_object_put_in_bucket_is_object(self):
|
|
self.storage.put(self.bucket_name, "file", b"content")
|
|
data_received = self.storage.get(self.bucket_name, "file")
|
|
assert b"content" == data_received
|
|
|
|
def test_getting_objects_put_in_bucket_are_objects(self):
|
|
self.storage.put(self.bucket_name, "file1", b"content 1")
|
|
self.storage.put(self.bucket_name, "file2", b"content 2")
|
|
data_received = self.storage.get_all(self.bucket_name)
|
|
assert {b"content 1", b"content 2"} == {*data_received}
|
|
|
|
|
|
def get_client(client_name):
|
|
if client_name == "mock":
|
|
return StorageAdapterMock(StorageClientMock())
|
|
if client_name == "azure":
|
|
return get_azure_storage()
|
|
|
|
|
|
@pytest.fixture
|
|
def storage_tester(client_name):
|
|
client = get_client(client_name)
|
|
return StorageTester(client)
|
|
|
|
|
|
@pytest.mark.parametrize("client_name", ["mock", "azure"])
|
|
def test_storages(storage_tester):
|
|
storage_tester.test_purging_bucket_yields_empty_bucket()
|
|
storage_tester.test_getting_objects_put_in_bucket_are_objects()
|
|
storage_tester.test_getting_object_put_in_bucket_is_object()
|
|
assert True
|