Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

azure tests pass

Browse Source
This commit is contained in:
Matthias Bisping 2022-02-24 10:19:18 +01:00
parent 8ea5eb6e00
commit e16590af20
2 changed files with 17 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
pyinfra/storage/adapters/azure.py
View File

@ -43,15 +43,15 @@ class AzureStorageAdapter(StorageAdapter):
blob_client.upload_blob(data, overwrite=True)
def get(self, bucket_name, object_name):
pass
# container_client = self.__provide_container_client(bucket_name)
# blob_client = container_client.get_blob_client(object_name)
# blob_data = blob_client.download_blob()
# return blob_data
container_client = self.__provide_container_client(bucket_name)
blob_client = container_client.get_blob_client(object_name)
blob_data = blob_client.download_blob()
return blob_data.readall()
def get_all(self, bucket_name):
container_client = self.__provide_container_client(bucket_name)
blobs = container_client.list_blobs(bucket_name)
blobs = container_client.list_blobs()
for blob in blobs:
logger.debug(f"Downloading '{blob.name}'...")
blob_client = container_client.get_blob_client(blob)

23
pyinfra/test/unit_tests/storage_test.py
View File

@ -5,6 +5,7 @@ import pytest
from pyinfra.storage.adapters.azure import AzureStorageAdapter
from pyinfra.storage.clients.azure import get_azure_client
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
@ -21,24 +22,22 @@ class StorageTester:
assert not {*data_received}
def test_getting_object_put_in_bucket_is_object(self):
with tempfile.SpooledTemporaryFile() as f:
self.storage.put(self.bucket_name, "file", f)
data_received = self.storage.get(self.bucket_name, "file")
assert f == data_received
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):
with tempfile.SpooledTemporaryFile() as f1, tempfile.SpooledTemporaryFile() as f2:
self.storage.put(self.bucket_name, "file1", f1)
self.storage.put(self.bucket_name, "file2", f2)
data_received = self.storage.get_all(self.bucket_name)
assert {f1, f2} == {*data_received}
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 AzureStorageAdapter(get_azure_client())
return get_azure_storage()
@pytest.fixture
@ -49,7 +48,7 @@ def storage_tester(client_name):
@pytest.mark.parametrize('client_name', ["mock", "azure"])
def test_storages(storage_tester):
# storage_tester.test_purging_bucket_yields_empty_bucket()
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()
storage_tester.test_getting_object_put_in_bucket_is_object()
assert True