Brian Bjarke JensenandCursor
39b104383c
PR Title Check / check-title (pull_request) Successful in 7s
Test Python Package / coverage-report (pull_request) Successful in 12s
Code Quality Pipeline / code-quality (pull_request) Successful in 48s
Test Python Package / unit-tests (pull_request) Successful in 19s
Test Python Package / integration-tests (pull_request) Successful in 28s
Drop bucket_name assertions that mypy treated as always false after manually clearing _bucket_name. Co-authored-by: Cursor <[email protected]>
94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
"""Unit tests for MinioAdapter instantiation and injection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from minio import Minio
|
|
|
|
from python_repositories.adapters.minio_adapter import MinioAdapter
|
|
from python_repositories.interfaces import ObjectRepositoryInterface
|
|
from tests.conftest import TEST_MINIO_CONFIG
|
|
|
|
|
|
def test_should_adhere_to_interface() -> None:
|
|
assert issubclass(MinioAdapter, ObjectRepositoryInterface)
|
|
_ = MinioAdapter(config=TEST_MINIO_CONFIG)
|
|
|
|
|
|
def test_should_have_logger_when_instantiated() -> None:
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG)
|
|
assert hasattr(adapter, "logger")
|
|
assert adapter.logger is not None
|
|
|
|
|
|
def test_should_not_be_connected_when_instantiated() -> None:
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG)
|
|
assert not adapter.is_connected()
|
|
|
|
|
|
def test_constructs_with_injected_config_without_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
for var in (
|
|
"MINIO_ENDPOINT",
|
|
"MINIO_ACCESS_KEY",
|
|
"MINIO_SECRET_KEY",
|
|
"MINIO_BUCKET",
|
|
):
|
|
monkeypatch.delenv(var, raising=False)
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG)
|
|
assert adapter._config == TEST_MINIO_CONFIG
|
|
|
|
|
|
def test_injected_client_sets_bucket_name() -> None:
|
|
mock_client = MagicMock(spec=Minio)
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
|
|
assert adapter._bucket_name == "test-bucket"
|
|
assert adapter._client is mock_client
|
|
|
|
|
|
def test_raises_when_client_provided_without_config() -> None:
|
|
mock_client = MagicMock(spec=Minio)
|
|
with pytest.raises(ValueError, match="config is required"):
|
|
MinioAdapter(client=mock_client)
|
|
|
|
|
|
def test_connect_with_injected_client_succeeds() -> None:
|
|
mock_client = MagicMock(spec=Minio)
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
|
|
|
|
adapter.connect()
|
|
|
|
mock_client.list_buckets.assert_called_once()
|
|
|
|
|
|
def test_connect_with_injected_client_raises_on_failure() -> None:
|
|
mock_client = MagicMock(spec=Minio)
|
|
mock_client.list_buckets.side_effect = Exception("connection lost")
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
|
|
|
|
with pytest.raises(ConnectionError, match="Could not connect to Minio"):
|
|
adapter.connect()
|
|
|
|
|
|
def test_connect_disconnects_before_reconnect(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
stale_client = MagicMock(spec=Minio)
|
|
new_client = MagicMock(spec=Minio)
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG)
|
|
adapter._client = stale_client
|
|
adapter._bucket_name = None
|
|
|
|
monkeypatch.setattr(
|
|
"python_repositories.adapters.minio_adapter.minio.Minio",
|
|
lambda *args, **kwargs: new_client,
|
|
)
|
|
|
|
adapter.connect()
|
|
|
|
assert adapter._client is new_client
|
|
new_client.list_buckets.assert_called_once()
|