142 lines
4.0 KiB
Python
142 lines
4.0 KiB
Python
"""Integration tests configuration."""
|
|
|
|
import os
|
|
from collections.abc import Generator
|
|
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)
|
|
def configure_logging() -> None:
|
|
"""Configure logging for the test session."""
|
|
# 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(),
|
|
wrapper_class=structlog.stdlib.BoundLogger,
|
|
cache_logger_on_first_use=True,
|
|
)
|
|
# Set up basic logging configuration
|
|
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."""
|
|
# Start container
|
|
container = RedisContainer(
|
|
image="redis/redis-stack:7.2.0-v0",
|
|
)
|
|
container.start()
|
|
# Set environment variable for Redis URI
|
|
redis_host = container.get_container_host_ip()
|
|
redis_port = container.get_exposed_port(6379)
|
|
redis_uri = f"redis://{redis_host}:{redis_port}"
|
|
|
|
yield redis_uri
|
|
|
|
# Stop container
|
|
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
|
|
|
|
yield env_vars
|
|
|
|
# Cleanup
|
|
for key in env_vars:
|
|
_ = os.environ.pop(key, default=None)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def raw_redis_client(redis_container: str) -> Generator[redis.Redis]:
|
|
"""Provide a raw Redis client connected to the test Redis container."""
|
|
# Connect client
|
|
client = redis.Redis.from_url(
|
|
url=redis_container,
|
|
socket_connect_timeout=10,
|
|
)
|
|
|
|
yield client
|
|
|
|
# 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)
|