python-repositories/tests/integration/conftest.py

"""Integration tests configuration."""

from collections.abc import Generator
import logging
from typing import Any, cast

from minio import Minio
import pytest
import redis
import structlog
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget
from testcontainers.minio import MinioContainer

from python_repositories.config import MinioConfig, RedisConfig

REDIS_PORT = 6379

MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin"
MINIO_BUCKET = "test-bucket"


class _RedisPingWaitStrategy(WaitStrategy):
    def __init__(self) -> None:
        super().__init__()
        self.with_transient_exceptions(redis.exceptions.ConnectionError)

    def wait_until_ready(self, container: WaitStrategyTarget) -> None:
        redis_container = cast("RedisTestContainer", container)
        if not self._poll(lambda: redis_container.get_client().ping()):
            raise redis.exceptions.ConnectionError("Could not connect to Redis")


class RedisTestContainer(DockerContainer):
    """Redis container using wait strategies instead of the deprecated decorator."""

    def __init__(self, image: str, port: int = REDIS_PORT) -> None:
        super().__init__(image, _wait_strategy=_RedisPingWaitStrategy())
        self.port = port
        self.with_exposed_ports(self.port)

    def get_client(self, **kwargs: Any) -> redis.Redis:
        return redis.Redis(
            host=self.get_container_host_ip(),
            port=self.get_exposed_port(self.port),
            **kwargs,
        )


@pytest.fixture(scope="session", autouse=True)
def configure_logging() -> None:
    """Configure logging for the test session."""
    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(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    logging.basicConfig(level=logging.ERROR)


@pytest.fixture(scope="session")
def redis_container() -> Generator[str, None, None]:
    """Set up a Redis container for testing and yield the Redis URI."""
    container = RedisTestContainer(
        image="redis/redis-stack:7.2.0-v0",
    )
    container.start()
    redis_host = container.get_container_host_ip()
    redis_port = container.get_exposed_port(REDIS_PORT)
    redis_uri = f"redis://{redis_host}:{redis_port}"

    yield redis_uri

    container.stop()


@pytest.fixture(scope="session")
def minio_container() -> Generator[dict[str, str], None, None]:
    """Set up a Minio container for testing and yield connection settings."""
    container = MinioContainer(
        image="minio/minio:latest",
        access_key=MINIO_ACCESS_KEY,
        secret_key=MINIO_SECRET_KEY,
    )
    container.start()
    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,
        "MINIO_SECURE": "false",
    }

    yield env_vars

    container.stop()


@pytest.fixture(scope="session")
def redis_config(redis_container: str) -> RedisConfig:
    """Provide RedisConfig built from the test container."""
    return RedisConfig(uri=redis_container)


@pytest.fixture(scope="session")
def minio_config(minio_container: dict[str, str]) -> MinioConfig:
    """Provide MinioConfig built from the test container."""
    return MinioConfig(
        endpoint=minio_container["MINIO_ENDPOINT"],
        access_key=minio_container["MINIO_ACCESS_KEY"],
        secret_key=minio_container["MINIO_SECRET_KEY"],
        bucket=minio_container["MINIO_BUCKET"],
        secure=False,
    )


@pytest.fixture(scope="session")
def raw_redis_client(redis_container: str) -> Generator[redis.Redis, None, None]:
    """Provide a raw Redis client connected to the test Redis container."""
    client = redis.Redis.from_url(
        url=redis_container,
        socket_connect_timeout=10,
    )

    yield client

    client.flushall()
    client.close()


@pytest.fixture(scope="session")
def raw_minio_client(minio_container: dict[str, str]) -> Generator[Minio, None, None]:
    """Provide a raw Minio client connected to the test Minio container."""
    client = Minio(
        endpoint=minio_container["MINIO_ENDPOINT"],
        access_key=minio_container["MINIO_ACCESS_KEY"],
        secret_key=minio_container["MINIO_SECRET_KEY"],
        secure=False,
    )
    bucket_name = minio_container["MINIO_BUCKET"]
    if not client.bucket_exists(bucket_name):
        client.make_bucket(bucket_name)

    yield client

    objects = client.list_objects(bucket_name, recursive=True)
    for obj in objects:
        client.remove_object(bucket_name, obj.object_name)