Brian Bjarke JensenandCursor
8c34534187
Test Python Package / integration-tests (pull_request) Successful in 27s
Code Quality Pipeline / code-quality (pull_request) Successful in 31s
Test Python Package / unit-tests (pull_request) Successful in 36s
Test Python Package / coverage-report (pull_request) Successful in 53s
PR Title Check / check-title (pull_request) Failing after 5s
Close and release the HTTP response in a finally block per SDK guidance, and add unit tests for success and read-failure cleanup. Co-authored-by: Cursor <[email protected]>
127 lines
4.0 KiB
Python
127 lines
4.0 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()
|
|
|
|
|
|
def test_get_closes_response_on_success() -> None:
|
|
"""get() must close and release the get_object HTTP response."""
|
|
mock_client = MagicMock(spec=Minio)
|
|
mock_client.bucket_exists.return_value = True
|
|
mock_response = MagicMock()
|
|
mock_response.read.side_effect = [b"data", b""]
|
|
mock_client.get_object.return_value = mock_response
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
|
|
|
|
result = adapter.get("some-object")
|
|
|
|
assert result is not None
|
|
assert result.read() == b"data"
|
|
mock_response.close.assert_called_once()
|
|
mock_response.release_conn.assert_called_once()
|
|
|
|
|
|
def test_get_closes_response_when_read_fails() -> None:
|
|
"""get() must close and release the response even if read() raises."""
|
|
mock_client = MagicMock(spec=Minio)
|
|
mock_client.bucket_exists.return_value = True
|
|
mock_response = MagicMock()
|
|
mock_response.read.side_effect = OSError("connection reset")
|
|
mock_client.get_object.return_value = mock_response
|
|
adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
|
|
|
|
result = adapter.get("some-object")
|
|
|
|
assert result is None
|
|
mock_response.close.assert_called_once()
|
|
mock_response.release_conn.assert_called_once()
|