81 lines
2.1 KiB
Python
81 lines
2.1 KiB
Python
"""Integration tests configuration."""
|
|
|
|
import os
|
|
from collections.abc import Generator
|
|
import pytest
|
|
import redis
|
|
import structlog
|
|
import logging
|
|
|
|
from testcontainers.redis import RedisContainer
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def configure_logging() -> None:
|
|
"""Configure logging for the test session."""
|
|
# Configure structlog
|
|
structlog.configure(
|
|
processors=[
|
|
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]:
|
|
"""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
|
|
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", autouse=True)
|
|
def set_environment_variables(
|
|
redis_container: str,
|
|
) -> Generator[dict[str, str]]:
|
|
"""Set environment variables needed for tests."""
|
|
# Build environment variables dictionary
|
|
env_vars = {"REDIS_URI": redis_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()
|