python-repositories/tests/integration/minio_adapter_test.py

"""Integration tests for the MinioAdapter."""

from collections.abc import Generator
from dataclasses import replace
import logging
import random
from io import BytesIO
from unittest.mock import MagicMock

import pytest
from minio import Minio, S3Error
from urllib3.response import BaseHTTPResponse

from python_repositories.adapters.minio_adapter import MinioAdapter
from python_repositories.config import MinioConfig
from tests.conftest import TEST_MINIO_CONFIG

pytestmark = pytest.mark.integration


def same_data(
    data_a: BytesIO,
    data_b: BytesIO,
) -> bool:
    """Check if two BytesIO-objects contain the same data."""
    data_a.seek(0)
    data_b.seek(0)
    data_a_bytes = data_a.read()
    data_b_bytes = data_b.read()
    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
    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, None, None]:
    """Provide a sample data bytes for tests."""
    random_bytes = random.randbytes(2**21)
    yield BytesIO(random_bytes)


@pytest.fixture(scope="function")
def data_in_minio(
    raw_minio_client: Minio,
    minio_config: MinioConfig,
    data: BytesIO,
) -> Generator[tuple[str, BytesIO], None, None]:
    """Fixture to set up a known value in Minio before each test."""
    object_name = "test_object"
    bucket_name = minio_config.bucket
    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,
    )
    data.seek(0)

    yield object_name, data

    raw_minio_client.remove_object(bucket_name, object_name)


@pytest.fixture(scope="module")
def minio_adapter(minio_config: MinioConfig) -> Generator[MinioAdapter, None, None]:
    """Fixture to provide a connected MinioAdapter instance."""
    adapter = MinioAdapter(config=minio_config)
    adapter.connect()
    yield adapter
    adapter.disconnect()


@pytest.fixture(scope="function", autouse=True)
def clear_minio(
    raw_minio_client: Minio,
    minio_config: MinioConfig,
) -> None:
    """Fixture to clear all Minio objects before each test."""
    bucket_name = minio_config.bucket
    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_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" in caplog.text
    assert "Minio" in caplog.text


def test_should_raise_connection_error_when_unable_to_connect() -> None:
    """Test that the MinioAdapter raises a ConnectionError when unable to connect."""
    config = MinioConfig(
        endpoint="invalid_uri",
        access_key="minioadmin",
        secret_key="minioadmin",
        bucket="test-bucket",
    )
    adapter = MinioAdapter(config=config)
    with pytest.raises(ConnectionError):
        adapter.connect()
    assert not adapter.is_connected()


def test_connect_raises_when_bucket_missing(
    raw_minio_client: Minio,
    minio_config: MinioConfig,
) -> None:
    """Test that connect fails when the configured bucket is missing."""
    bucket_name = minio_config.bucket
    raw_minio_client.remove_bucket(bucket_name)
    adapter = MinioAdapter(config=minio_config)

    try:
        with pytest.raises(ConnectionError, match="does not exist"):
            adapter.connect()
        assert not adapter.is_connected()
    finally:
        raw_minio_client.make_bucket(bucket_name)


def test_connect_creates_bucket_when_create_bucket_if_missing_enabled(
    raw_minio_client: Minio,
    minio_config: MinioConfig,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test that connect can create the configured bucket when enabled."""
    bucket_name = minio_config.bucket
    raw_minio_client.remove_bucket(bucket_name)
    config = replace(minio_config, create_bucket_if_missing=True)
    adapter = MinioAdapter(config=config)

    with caplog.at_level(logging.INFO):
        adapter.connect()

    assert "Creating bucket" in caplog.text
    assert bucket_name in caplog.text


def test_should_log_error_on_exception_during_exit(
    minio_config: MinioConfig,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test that the MinioAdapter logs an error if an exception occurs during __exit__."""
    try:
        with MinioAdapter(config=minio_config) as adapter:
            assert adapter.is_connected()
            raise ValueError("Simulated error")
    except ValueError:
        pass
    assert "Error while exiting context" in caplog.text


def test_should_have_context_manager(minio_config: MinioConfig) -> None:
    """Test that the MinioAdapter can be used as a context manager."""
    with MinioAdapter(config=minio_config) 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."""
    object_name, expected_data = data_in_minio
    received_data = minio_adapter.get(object_name)
    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."""
    received_data = minio_adapter.get("nonexistent_object")
    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."""
    invalid_object_names = ["", 123, None]
    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_config: MinioConfig,
) -> None:
    """Test that the MinioAdapter raises ConnectionError when getting while not connected."""
    adapter = MinioAdapter(config=minio_config)
    with pytest.raises(ConnectionError):
        adapter.get("some_object")


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."""
    mock_client = MagicMock(spec=Minio)
    mock_client.bucket_exists.return_value = True
    mock_client.get_object.side_effect = S3Error(
        MagicMock(spec=BaseHTTPResponse),
        "NoSuchKey",
        "",
        "",
        "",
        "",
        bucket_name="test-bucket",
        object_name="missing-object",
    )
    adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
    object_name = "missing-object"
    with caplog.at_level("WARNING"):
        result = adapter.get(object_name)
    assert result is None
    assert "Object not found" in caplog.text
    assert object_name in caplog.text
    assert adapter._bucket_name in caplog.text


def test_should_reraise_s3error_other_than_no_such_key() -> None:
    """Test that the MinioAdapter re-raises unhandled S3 errors."""
    mock_client = MagicMock(spec=Minio)
    mock_client.bucket_exists.return_value = True
    other_s3error = S3Error(
        MagicMock(spec=BaseHTTPResponse),
        "UnhandledError",
        "",
        "",
        "",
        "",
        bucket_name="test-bucket",
        object_name="missing-object",
    )
    mock_client.get_object.side_effect = other_s3error
    adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
    with pytest.raises(S3Error) as exc_info:
        adapter.get("missing-object")
    assert exc_info.value.code == "UnhandledError"


def test_should_reraise_general_exception() -> None:
    """Test that the MinioAdapter re-raises general exceptions during get."""
    mock_client = MagicMock(spec=Minio)
    mock_client.bucket_exists.return_value = True
    general_exception = Exception("General failure")
    mock_client.get_object.side_effect = general_exception
    adapter = MinioAdapter(config=TEST_MINIO_CONFIG, client=mock_client)
    with pytest.raises(Exception, match="General failure"):
        adapter.get("missing-object")


def test_should_put_data(
    data: BytesIO,
    minio_adapter: MinioAdapter,
    minio_config: MinioConfig,
) -> None:
    """Test that the MinioAdapter can put data into a bucket."""
    object_name = "new_test_object"
    received_data = minio_adapter.get(object_name)
    assert received_data is None
    minio_adapter.put(object_name, data)
    received_data = minio_adapter.get(object_name)
    assert received_data is not None
    assert same_data(data, received_data)
    minio_adapter._client.remove_object(minio_config.bucket, object_name)  # type: ignore[union-attr]


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."""
    object_name, _ = data_in_minio
    new_data = BytesIO(random.randbytes(2**21))
    received_data = minio_adapter.get(object_name)
    assert received_data is not None
    assert not same_data(received_data, new_data)
    minio_adapter.put(object_name, new_data)
    received_data = minio_adapter.get(object_name)
    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."""
    invalid_object_names = ["", 123, None]
    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."""
    object_name = "valid_object_name"
    invalid_data = ["not_bytesio", 123, None]
    for invalid in invalid_data:
        with pytest.raises(ValueError):
            minio_adapter.put(object_name, invalid)  # type: ignore


def test_should_raise_value_error_on_invalid_put_content_type(
    data: BytesIO,
    minio_adapter: MinioAdapter,
) -> None:
    """Test that the MinioAdapter raises ValueError when putting with an invalid content type."""
    object_name = "valid_object_name"
    invalid_content_types = ["", 123, None]
    for content_type in invalid_content_types:
        with pytest.raises(ValueError):
            minio_adapter.put(object_name, data, content_type)  # type: ignore


def test_should_raise_connection_error_on_put_when_not_connected(
    minio_config: MinioConfig,
    data: BytesIO,
) -> None:
    """Test that the MinioAdapter raises ConnectionError when putting while not connected."""
    adapter = MinioAdapter(config=minio_config)
    with pytest.raises(ConnectionError):
        adapter.put("some_object", data)


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."""
    object_name, _ = data_in_minio
    received_data = minio_adapter.get(object_name)
    assert received_data is not None
    minio_adapter.delete(object_name)
    assert minio_adapter.get(object_name) 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."""
    invalid_object_names = ["", 123, None]
    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_config: MinioConfig,
) -> None:
    """Test that the MinioAdapter raises ConnectionError when deleting while not connected."""
    adapter = MinioAdapter(config=minio_config)
    with pytest.raises(ConnectionError):
        adapter.delete("some_object")


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."""
    object_name, _ = data_in_minio
    new_data = BytesIO(random.randbytes(2**21))
    new_data_name = "another_test_object"
    minio_adapter.put(new_data_name, new_data)
    objects = minio_adapter.list_objects()
    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."""
    object_name, _ = data_in_minio
    new_data = BytesIO(random.randbytes(2**21))
    new_data_name = "prefix_test_object"
    minio_adapter.put(new_data_name, new_data)
    prefix = "prefix_"
    objects = minio_adapter.list_objects(prefix)
    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."""
    invalid_prefixes = [123, None]
    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_config: MinioConfig,
) -> None:
    """Test that the MinioAdapter raises ConnectionError when listing while not connected."""
    adapter = MinioAdapter(config=minio_config)
    with pytest.raises(ConnectionError):
        adapter.list_objects()


if __name__ == "__main__":
    pytest.main(["-s", "-v", __file__])