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

RED-3463: Add some important tests

Browse Source
This commit is contained in:
Julius Unverfehrt 2022-03-04 13:05:27 +01:00
parent 666bb067ae
commit aa99d15c04
3 changed files with 37 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
pyinfra/test/unit_tests/conftest.py
View File

@ -1,8 +1,22 @@
import json import json
import logging
from unittest.mock import Mock from unittest.mock import Mock
import pytest import pytest
from pyinfra.exceptions import UnknownClient
from pyinfra.locations import TEST_DIR from pyinfra.locations import TEST_DIR
from pyinfra.storage.adapters.azure import AzureStorageAdapter
from pyinfra.storage.adapters.s3 import S3StorageAdapter
from pyinfra.storage.clients.azure import get_azure_client
from pyinfra.storage.clients.s3 import get_s3_client
from pyinfra.storage.storage import Storage
from pyinfra.test.config import CONFIG
from pyinfra.test.storage.adapter_mock import StorageAdapterMock
from pyinfra.test.storage.client_mock import StorageClientMock
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@pytest.fixture @pytest.fixture
@ -35,3 +49,25 @@ def mock_make_load_data():
return storage_data return storage_data
return load_data return load_data
@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)
yield storage
logger.debug("Teardown for storage")
storage.clear_bucket(bucket_name)
def get_adapter(client_name, s3_backend):
if client_name == "mock":
return StorageAdapterMock(StorageClientMock())
if client_name == "azure":
return AzureStorageAdapter(get_azure_client(CONFIG.storage.azure.connection_string))
if client_name == "s3":
return S3StorageAdapter(get_s3_client(CONFIG.storage[s3_backend]))
else:
raise UnknownClient(client_name)

3
pyinfra/test/unit_tests/core_test.py
View File

@ -1,6 +1,5 @@
import json import json
from time import sleep from unittest.mock import patch
from unittest.mock import patch, Mock, MagicMock
from pyinfra.core import make_analyzer, make_payload_processor from pyinfra.core import make_analyzer, make_payload_processor
from pyinfra.test.config import CONFIG from pyinfra.test.config import CONFIG

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

@ -2,16 +2,8 @@ import logging
import pytest import pytest
from pyinfra.exceptions import UnknownClient
from pyinfra.storage.adapters.azure import AzureStorageAdapter
from pyinfra.storage.adapters.s3 import S3StorageAdapter
from pyinfra.storage.clients.azure import get_azure_client
from pyinfra.storage.clients.s3 import get_s3_client
from pyinfra.storage.storage import Storage from pyinfra.storage.storage import Storage
from pyinfra.storage.storages import get_azure_storage, get_s3_storage from pyinfra.storage.storages import get_azure_storage, get_s3_storage
from pyinfra.test.config import CONFIG
from pyinfra.test.storage.adapter_mock import StorageAdapterMock
from pyinfra.test.storage.client_mock import StorageClientMock
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
@ -51,28 +43,6 @@ class TestStorage:
assert {(bucket_name, "file1"), (bucket_name, "file2")} == {*full_names_received} assert {(bucket_name, "file1"), (bucket_name, "file2")} == {*full_names_received}
def get_adapter(client_name, s3_backend):
if client_name == "mock":
return StorageAdapterMock(StorageClientMock())
if client_name == "azure":
return AzureStorageAdapter(get_azure_client(CONFIG.storage.azure.connection_string))
if client_name == "s3":
return S3StorageAdapter(get_s3_client(CONFIG.storage[s3_backend]))
else:
raise UnknownClient(client_name)
@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)
yield storage
logger.debug("Teardown for storage")
storage.clear_bucket(bucket_name)
def test_get_azure_storage_yields_storage(): def test_get_azure_storage_yields_storage():
assert isinstance(get_azure_storage(), Storage) assert isinstance(get_azure_storage(), Storage)