From a3b5443e0dcf2715893a4ea3b5114a0a0bd19881 Mon Sep 17 00:00:00 2001 From: Brian Bjarke Jensen Date: Mon, 15 Sep 2025 00:43:43 +0200 Subject: [PATCH] added minio adapter and tests --- python_repositories/adapters/minio_adapter.py | 202 ++++++++ tests/integration/conftest.py | 65 ++- tests/integration/minio_adapter_test.py | 467 ++++++++++++++++++ 3 files changed, 732 insertions(+), 2 deletions(-) create mode 100644 python_repositories/adapters/minio_adapter.py create mode 100644 tests/integration/minio_adapter_test.py diff --git a/python_repositories/adapters/minio_adapter.py b/python_repositories/adapters/minio_adapter.py new file mode 100644 index 0000000..51ea42b --- /dev/null +++ b/python_repositories/adapters/minio_adapter.py @@ -0,0 +1,202 @@ +"""Definition of MinioAdapter class.""" + +from __future__ import annotations +import os +from io import BytesIO + +import structlog +from minio import Minio, S3Error + +from python_utils import check_env + +from python_repositories.interfaces import ( + ContextAwareInterface, + ConnectionAwareInterface, +) + + +class MinioAdapter( + ContextAwareInterface, + ConnectionAwareInterface, +): + """Minio adapter exposing basic CRUD functionality.""" + + endpoint_env_var_name: str = "MINIO_ENDPOINT" + access_key_env_var_name: str = "MINIO_ACCESS_KEY" + secret_key_env_var_name: str = "MINIO_SECRET_KEY" + bucket_env_var_name: str = "MINIO_BUCKET" + chunk_size: int = 5*2**20 # 5 MiB + + def __init__(self) -> None: + # Setup logger + self.logger = structlog.get_logger( + self.__class__.__name__, + ) + # Check environment variables + check_env( + { + self.endpoint_env_var_name, + self.access_key_env_var_name, + self.secret_key_env_var_name, + self.bucket_env_var_name, + }, + ) + # Prepare internal variables + self._client: Minio | None = None + self._bucket_name: str | None = None + + def __enter__(self) -> MinioAdapter: + """Enter the context.""" + self.connect() + return self + + def __exit__( + self, exc_type: type | None, exc_val: object | None, exc_tb: object | None + ) -> None: + """Exit the context.""" + ctx_info = {"exc_type": exc_type, "exc_val": exc_val, "exc_tb": exc_tb} + if any( + ( + exc_type is not None, + exc_val is not None, + exc_tb is not None, + ), + ): + self.logger.error("Error while exiting context", **ctx_info) + self.disconnect() + + def connect(self) -> None: + """Connect to the Minio server.""" + # Stop if already connected + if self.is_connected: + self.logger.info("Already connected to Minio") + return + # Prepare arguments + endpoint = str(os.getenv(self.endpoint_env_var_name)) + access_key = str(os.getenv(self.access_key_env_var_name)) + secret_key = str(os.getenv(self.secret_key_env_var_name)) + bucket = str(os.getenv(self.bucket_env_var_name)) + # Connect client + client = Minio( + endpoint=endpoint, + access_key=access_key, + secret_key=secret_key, + secure=False, + ) + # Test the connection by listing buckets (will raise if connection fails) + try: + _ = client.list_buckets() + except Exception as exc: # pylint: disable=broad-except + raise ConnectionError(f"Could not connect to Minio at {endpoint}") from exc + # Ensure bucket exists + if not client.bucket_exists(bucket): + self.logger.info(f"Creating bucket '{bucket}'") + client.make_bucket(bucket) + # Persist information + self._client = client + self._bucket_name = bucket + + def disconnect(self) -> None: + """Disconnect from the Minio server.""" + # Close connection + # N.B. Minio client does not have a close method, but we include this for symmetry with other adapters + # Reset client + self._client = None + self._bucket_name = None + + @property + def is_connected(self) -> bool: + """Check if connected to Minio server.""" + res = bool(isinstance(self._client, Minio)) + self.logger.debug(res) + return res + + def _put(self, object_name: str, data: BytesIO) -> None: + """Put an object into the Minio bucket.""" + # Check input + if not isinstance(object_name, str) or len(object_name) == 0: + raise ValueError("object_name must be a non-empty string") + if not isinstance(data, BytesIO) or data.getbuffer().nbytes == 0: + raise ValueError("data must be a non-empty BytesIO object") + # Check connection + if self._client is None or not self.is_connected: + raise ConnectionError("Not connected to Minio") + # Prepare buffer for reading + num_bytes = data.getbuffer().nbytes + data.seek(0) + # Send data to bucket + # N.B. bucket name is set when connecting + self._client.put_object( + bucket_name=self._bucket_name, # type: ignore + object_name=object_name, + data=data, + length=num_bytes, + part_size=self.chunk_size, + ) + self.logger.debug(f"Put object '{object_name}' into bucket '{self._bucket_name}'") + + def _get(self, object_name: str) -> BytesIO | None: + """Get an object from the Minio bucket.""" + # Check input + if not isinstance(object_name, str) or len(object_name) == 0: + raise ValueError("object_name must be a non-empty string") + # Check connection + if self._client is None or not self.is_connected: + raise ConnectionError("Not connected to Minio") + # Get data from bucket + # N.B. bucket name is set when connecting + try: + response = self._client.get_object( + bucket_name=self._bucket_name, # type: ignore + object_name=object_name, + ) + # Get buffered data + buffer = BytesIO() + while chunk := response.read(self.chunk_size): + buffer.write(chunk) + buffer.seek(0) + self.logger.debug(f"Got object '{object_name}' from bucket '{self._bucket_name}'") + return buffer + except S3Error as exc: + if exc.code == "NoSuchKey": + self.logger.warning(f"Object '{object_name}' not found in bucket '{self._bucket_name}'") + else: + self.logger.error(repr(exc)) + except Exception as exc: # pylint: disable=broad-except + self.logger.error(repr(exc)) + return None + + def _delete(self, object_name: str) -> None: + """Delete an object from the Minio bucket.""" + # Check input + if not isinstance(object_name, str) or len(object_name) == 0: + raise ValueError("object_name must be a non-empty string") + # Check connection + if self._client is None or not self.is_connected: + raise ConnectionError("Not connected to Minio") + # Delete object from bucket + # N.B. bucket name is set when connecting + self._client.remove_object( + bucket_name=self._bucket_name, # type: ignore + object_name=object_name, + ) + self.logger.debug(f"Deleted object '{object_name}' from bucket '{self._bucket_name}'") + + def _list_objects(self, prefix: str = "") -> list[str]: + """List objects in the Minio bucket with an optional prefix.""" + # Check input + if not isinstance(prefix, str): + raise ValueError("prefix must be a string") + # Check connection + # N.B. bucket name is set when connecting + if self._client is None or not self.is_connected: + raise ConnectionError("Not connected to Minio") + # List objects in bucket + objects = self._client.list_objects( + bucket_name=self._bucket_name, # type: ignore + prefix=prefix, + recursive=True, + ) + object_names = [obj.object_name for obj in objects if obj.object_name is not None] + self.logger.debug(f"Listed {len(object_names)} object(s) in bucket '{self._bucket_name}' with prefix '{prefix}'") + return object_names diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index e7547e5..208357c 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -6,8 +6,14 @@ import pytest import redis import structlog import logging +from minio import Minio from testcontainers.redis import RedisContainer +from testcontainers.minio import MinioContainer + +MINIO_ACCESS_KEY = "minioadmin" +MINIO_SECRET_KEY = "minioadmin" +MINIO_BUCKET = "test-bucket" @pytest.fixture(scope="session", autouse=True) @@ -16,6 +22,10 @@ def configure_logging() -> None: # Configure structlog structlog.configure( processors=[ + structlog.stdlib.filter_by_level, + structlog.stdlib.add_logger_name, + structlog.stdlib.add_log_level, + structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], logger_factory=structlog.stdlib.LoggerFactory(), @@ -27,12 +37,11 @@ def configure_logging() -> None: @pytest.fixture(scope="session") -def redis_container() -> Generator[str]: +def redis_container() -> Generator[str, None, None]: """Set up a Redis container for testing and yield the Redis URI.""" # Start container container = RedisContainer( image="redis/redis-stack:7.2.0-v0", - port=6379, ) container.start() # Set environment variable for Redis URI @@ -46,13 +55,42 @@ def redis_container() -> Generator[str]: container.stop() +@pytest.fixture(scope="session") +def minio_container() -> Generator[dict[str, str], None, None]: + """Set up a Minio container for testing and yield the Minio URI.""" + # Start container + container = MinioContainer( + image="minio/minio:latest", + access_key=MINIO_ACCESS_KEY, + secret_key=MINIO_SECRET_KEY, + ) + container.start() + # Build environment variables dictionary + minio_host = container.get_container_host_ip() + minio_port = container.get_exposed_port(9000) + minio_endpoint = f"{minio_host}:{minio_port}" + env_vars = { + "MINIO_ENDPOINT": minio_endpoint, + "MINIO_ACCESS_KEY": MINIO_ACCESS_KEY, + "MINIO_SECRET_KEY": MINIO_SECRET_KEY, + "MINIO_BUCKET": MINIO_BUCKET, + } + + yield env_vars + + # Stop container + container.stop() + + @pytest.fixture(scope="session", autouse=True) def set_environment_variables( redis_container: str, + minio_container: dict[str, str], ) -> Generator[dict[str, str]]: """Set environment variables needed for tests.""" # Build environment variables dictionary env_vars = {"REDIS_URI": redis_container} + env_vars.update(minio_container) # Set environment variables for key, value in env_vars.items(): os.environ[key] = value @@ -78,3 +116,26 @@ def raw_redis_client(redis_container: str) -> Generator[redis.Redis]: # Cleanup client.flushall() client.close() + + +@pytest.fixture(scope="session") +def raw_minio_client(minio_container: dict[str, str]) -> Generator[Minio]: + """Provide a raw Minio client connected to the test Minio container.""" + # Connect client + client = Minio( + endpoint=minio_container["MINIO_ENDPOINT"], + access_key=minio_container["MINIO_ACCESS_KEY"], + secret_key=minio_container["MINIO_SECRET_KEY"], + secure=False, + ) + # Ensure bucket exists + bucket_name = minio_container["MINIO_BUCKET"] + if not client.bucket_exists(bucket_name): + client.make_bucket(bucket_name) + + yield client + + # Cleanup + objects = client.list_objects(bucket_name, recursive=True) + for obj in objects: + client.remove_object(bucket_name, obj.object_name) \ No newline at end of file diff --git a/tests/integration/minio_adapter_test.py b/tests/integration/minio_adapter_test.py new file mode 100644 index 0000000..47e3ca1 --- /dev/null +++ b/tests/integration/minio_adapter_test.py @@ -0,0 +1,467 @@ +# pylint: disable=protected-access +# The above line disables pylint's protected member access warnings for this file, +# allowing tests to access MinioAdapter's internal methods as needed for integration testing. +"""Integration tests for the MinioAdapter.""" + +from collections.abc import Generator +from minio import S3Error +import pytest +from unittest.mock import MagicMock, patch +from minio import Minio +from io import BytesIO +import random +import os +import logging +from python_repositories.adapters.minio_adapter import MinioAdapter + + +def same_data( + data_a: BytesIO, + data_b: BytesIO, +) -> bool: + """Check if two BytesIO-objects contain the same data.""" + assert isinstance(data_a, BytesIO) + assert isinstance(data_b, BytesIO) + # prepare for being read + data_a.seek(0) + data_b.seek(0) + # convert to bytes + data_a_bytes = data_a.read() + data_b_bytes = data_b.read() + # compare size + if len(data_a_bytes) != len(data_b_bytes): + logging.error( + 'data has different length: %s and %s', + len(data_a_bytes), + len(data_b_bytes), + ) + return False + # compare content + if data_a_bytes != data_b_bytes: + logging.error('data has different bytes') + return False + return True + + +@pytest.fixture(scope="module") +def data() -> Generator[BytesIO]: + """Provide a sample data bytes for tests.""" + # Generate random bytes + random_bytes = random.randbytes(2**21) # 2 MiB + yield BytesIO(random_bytes) + + +@pytest.fixture(scope="function") +def data_in_minio( + raw_minio_client: Minio, + data: BytesIO, +) -> Generator[tuple[str, BytesIO]]: + """Fixture to set up a known value in Minio before each test.""" + object_name = "test_object" + bucket_name = str(os.getenv("MINIO_BUCKET")) + # Upload object + num_bytes = data.getbuffer().nbytes + data.seek(0) + raw_minio_client.put_object( + bucket_name, + object_name, + data, + length=num_bytes, + part_size=MinioAdapter.chunk_size, + ) + # Reset data for reading in tests + data.seek(0) + + yield object_name, data + + # Cleanup + raw_minio_client.remove_object(bucket_name, object_name) + + +@pytest.fixture(scope="module") +def minio_adapter() -> Generator[MinioAdapter]: + """Fixture to provide a connected MinioAdapter instance.""" + adapter = MinioAdapter() + adapter.connect() + yield adapter + adapter.disconnect() + + +@pytest.fixture(scope="function", autouse=True) +def clear_minio( + raw_minio_client: Minio, +) -> None: + """Fixture to clear all Minio objects before each test.""" + bucket_name = str(os.getenv("MINIO_BUCKET")) + # Clear all objects before each test + objects = raw_minio_client.list_objects(bucket_name, recursive=True) + for obj in objects: + if not obj.object_name: + continue + raw_minio_client.remove_object(bucket_name, obj.object_name) + + +def test_should_adhere_to_interface() -> None: + """Test that the MinioAdapter adheres to the expected interface.""" + # Instantiation fails if interface not adhered to + _ = MinioAdapter() + + +def test_should_have_logger_when_instantiated() -> None: + """Test that the MinioAdapter has a logger when instantiated.""" + adapter = MinioAdapter() + assert hasattr(adapter, "logger") + + +def test_should_not_be_connected_when_instantiated() -> None: + """Test that the MinioAdapter is not connected when instantiated.""" + adapter = MinioAdapter() + assert not adapter.is_connected + + +def test_should_log_info_when_already_connected( + minio_adapter: MinioAdapter, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs info when connect is called while already connected.""" + with caplog.at_level(logging.INFO): + minio_adapter.connect() + assert "Already connected to Minio" in caplog.text + + +def test_should_raise_connection_error_when_unable_to_connect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test that the MinioAdapter raises a ConnectionError when unable to connect.""" + # Arrange + monkeypatch.setenv("MINIO_ENDPOINT", "invalid_uri") + adapter = MinioAdapter() + with pytest.raises(ConnectionError): + adapter.connect() + assert not adapter.is_connected + + +def test_should_log_info_when_creating_expected_bucket( + raw_minio_client: Minio, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs info when creating the expected bucket.""" + # Arrange + bucket_name = str(os.getenv("MINIO_BUCKET")) + raw_minio_client.remove_bucket(bucket_name) + adapter = MinioAdapter() + # Act + with caplog.at_level(logging.INFO): + adapter.connect() + # Assert + assert f"Creating bucket '{bucket_name}'" in caplog.text + + +def test_should_log_error_on_exception_during_exit( + minio_container: dict[str, str], + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs an error if an exception occurs during __exit__.""" + try: + with MinioAdapter() as adapter: + assert adapter.is_connected + raise ValueError("Simulated error") + except ValueError: + pass # Expected + # Assert error was logged + assert "Error while exiting context" in caplog.text + + +def test_should_have_context_manager() -> None: + """Test that the MinioAdapter can be used as a context manager.""" + with MinioAdapter() as adapter: + assert adapter._client is not None + assert adapter._client is None + + +def test_should_get_data( + data_in_minio: tuple[str, BytesIO], + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can get data from a bucket.""" + # Arrange + object_name, expected_data = data_in_minio + # Act + received_data = minio_adapter._get(object_name) # type: ignore + # Assert + assert received_data is not None + assert same_data(expected_data, received_data) + + +def test_should_get_none_for_nonexistent_object( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter returns None for a nonexistent object.""" + # Arrange + object_name = "nonexistent_object" + # Act + received_data = minio_adapter._get(object_name) # type: ignore + # Assert + assert received_data is None + + +def test_should_raise_value_error_on_invalid_get_object_name( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ValueError when getting with an invalid object name.""" + # Arrange + invalid_object_names = ["", 123, None] + # Act & Assert + for object_name in invalid_object_names: + with pytest.raises(ValueError): + minio_adapter._get(object_name) # type: ignore + + +def test_should_raise_connection_error_on_get_when_not_connected( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ConnectionError when getting while not connected.""" + # Arrange + adapter = MinioAdapter() # not connected + # Act & Assert + with pytest.raises(ConnectionError): + adapter._get("some_object") # type: ignore + + +def test_should_log_warning_when_getting_nonexistent_object( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs a warning when getting a nonexistent object.""" + # Arrange + adapter = MinioAdapter() + adapter._client = MagicMock(spec=Minio) + adapter._client.get_object.side_effect = S3Error(code="NoSuchKey", message="", resource="", request_id="", host_id="", response="", bucket_name="test-bucket", object_name="missing-object") + adapter._bucket_name = "test-bucket" + object_name = "missing-object" + # Act + with caplog.at_level("WARNING"): + result = adapter._get(object_name) + # Assert + assert result is None + assert f"Object '{object_name}' not found in bucket '{adapter._bucket_name}'" in caplog.text + + +def test_should_log_error_when_getting_with_s3error_other_than_no_such_key( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs an error when getting a nonexistent object.""" + # Arrange + adapter = MinioAdapter() + adapter._client = MagicMock(spec=Minio) + other_s3error = S3Error(code="UnhandledError", message="", resource="", request_id="", host_id="", response="", bucket_name="test-bucket", object_name="missing-object") + adapter._client.get_object.side_effect = other_s3error + adapter._bucket_name = "test-bucket" + object_name = "missing-object" + # Act + with caplog.at_level("ERROR"): + result = adapter._get(object_name) + # Assert + assert result is None + assert repr(other_s3error) in caplog.text + + +def test_should_log_error_when_getting_with_general_exception( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that the MinioAdapter logs an error when getting a nonexistent object.""" + # Arrange + adapter = MinioAdapter() + adapter._client = MagicMock(spec=Minio) + general_exception = Exception("General failure") + adapter._client.get_object.side_effect = general_exception + adapter._bucket_name = "test-bucket" + object_name = "missing-object" + # Act + with caplog.at_level("ERROR"): + result = adapter._get(object_name) + # Assert + assert result is None + assert repr(general_exception) in caplog.text + + +def test_should_put_data( + data: BytesIO, + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can put data into a bucket.""" + # Arrange + object_name = "new_test_object" + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is None # ensure object does not exist yet + # Act + minio_adapter._put(object_name, data) # type: ignore + # Assert + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is not None + assert same_data(data, received_data) + # Cleanup + bucket_name = str(os.getenv("MINIO_BUCKET")) + minio_adapter._client.remove_object(bucket_name, object_name) + + +def test_should_update_data( + data_in_minio: tuple[str, BytesIO], + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can update data in a bucket.""" + # Arrange + object_name, _ = data_in_minio + new_data = BytesIO(random.randbytes(2**21)) # 2 MiB + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is not None + assert not same_data(received_data, new_data) + # Act + minio_adapter._put(object_name, new_data) # type: ignore + # Assert + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is not None + assert same_data(new_data, received_data) + + +def test_should_raise_value_error_on_invalid_put_object_name( + data: BytesIO, + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ValueError when putting with an invalid object name.""" + # Arrange + invalid_object_names = ["", 123, None] + # Act & Assert + for object_name in invalid_object_names: + with pytest.raises(ValueError): + minio_adapter._put(object_name, data) # type: ignore + + +def test_should_raise_value_error_on_invalid_put_data( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ValueError when putting with invalid data.""" + # Arrange + object_name = "valid_object_name" + invalid_data = ["not_bytesio", 123, None] + # Act & Assert + for data in invalid_data: + with pytest.raises(ValueError): + minio_adapter._put(object_name, data) # type: ignore + + +def test_should_raise_connection_error_on_put_when_not_connected( + data: BytesIO, + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ConnectionError when putting while not connected.""" + # Arrange + adapter = MinioAdapter() # not connected + # Act & Assert + with pytest.raises(ConnectionError): + adapter._put("some_object", data) # type: ignore + + +def test_should_delete_object( + data_in_minio: tuple[str, BytesIO], + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can delete an object from a bucket.""" + # Arrange + object_name, _ = data_in_minio + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is not None # ensure object exists + # Act + minio_adapter._delete(object_name) # type: ignore + # Assert + received_data = minio_adapter._get(object_name) # type: ignore + assert received_data is None + + +def test_should_raise_value_error_on_invalid_delete_object_name( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ValueError when deleting with an invalid object name.""" + # Arrange + invalid_object_names = ["", 123, None] + # Act & Assert + for object_name in invalid_object_names: + with pytest.raises(ValueError): + minio_adapter._delete(object_name) # type: ignore + + +def test_should_raise_connection_error_on_delete_when_not_connected( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ConnectionError when deleting while not connected.""" + # Arrange + adapter = MinioAdapter() # not connected + # Act & Assert + with pytest.raises(ConnectionError): + adapter._delete("some_object") # type: ignore + + +def test_should_list_objects( + data_in_minio: tuple[str, BytesIO], + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can list objects in a bucket.""" + # Arrange + object_name, _ = data_in_minio + new_data = BytesIO(random.randbytes(2**21)) # 2 MiB + new_data_name = "another_test_object" + minio_adapter._put(new_data_name, new_data) # type: ignore + # Act + objects = minio_adapter._list_objects() # type: ignore + # Assert + assert isinstance(objects, list) + assert len(objects) == 2 + assert object_name in objects + assert new_data_name in objects + + +def test_should_list_objects_with_prefix( + data_in_minio: tuple[str, BytesIO], + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter can list objects in a bucket with a prefix.""" + # Arrange + object_name, _ = data_in_minio + new_data = BytesIO(random.randbytes(2**21)) # 2 MiB + new_data_name = "prefix_test_object" + minio_adapter._put(new_data_name, new_data) # type: ignore + prefix = "prefix_" + # Act + objects = minio_adapter._list_objects(prefix) # type: ignore + # Assert + assert isinstance(objects, list) + assert len(objects) == 1 + assert new_data_name in objects + assert object_name not in objects + + +def test_should_raise_value_error_on_invalid_list_objects_prefix( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ValueError when listing with an invalid prefix.""" + # Arrange + invalid_prefixes = [123, None] + # Act & Assert + for prefix in invalid_prefixes: + with pytest.raises(ValueError): + minio_adapter._list_objects(prefix) # type: ignore + + +def test_should_raise_connection_error_on_list_objects_when_not_connected( + minio_adapter: MinioAdapter, +) -> None: + """Test that the MinioAdapter raises ConnectionError when listing while not connected.""" + # Arrange + adapter = MinioAdapter() # not connected + # Act & Assert + with pytest.raises(ConnectionError): + adapter._list_objects() # type: ignore + + +# allows local debugging by running file as script +if __name__ == "__main__": + pytest.main(["-s", "-v", __file__])