Brian Bjarke JensenandCursor
b28ac6e803
PR Title Check / check-title (pull_request) Successful in 7s
Test Python Package / unit-tests (pull_request) Successful in 13s
Test Python Package / integration-tests (pull_request) Successful in 23s
Test Python Package / coverage-report (pull_request) Successful in 10s
Code Quality Pipeline / code-quality (pull_request) Successful in 49s
Split Redis and MinIO container setup into backend-specific conftest modules so only the containers needed for collected tests are imported and started. Co-authored-by: Cursor <[email protected]>
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""Redis container session helpers for integration tests."""
|
|
|
|
from collections.abc import Generator
|
|
from typing import Any, cast
|
|
|
|
import redis
|
|
from testcontainers.core.container import DockerContainer
|
|
from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget
|
|
|
|
from python_repositories.config import RedisConfig
|
|
|
|
REDIS_PORT = 6379
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
_redis_uri: str | None = None
|
|
_redis_container: RedisTestContainer | None = None
|
|
_raw_redis_client: redis.Redis | None = None
|
|
_redis_uri_refs = 0
|
|
_raw_redis_client_refs = 0
|
|
|
|
|
|
def redis_uri() -> Generator[str, None, None]:
|
|
"""Yield a session-scoped Redis URI, starting the container once."""
|
|
global _redis_uri, _redis_container, _redis_uri_refs
|
|
if _redis_uri is None:
|
|
_redis_container = RedisTestContainer(image="redis/redis-stack:7.2.0-v0")
|
|
_redis_container.start()
|
|
redis_host = _redis_container.get_container_host_ip()
|
|
redis_port = _redis_container.get_exposed_port(REDIS_PORT)
|
|
_redis_uri = f"redis://{redis_host}:{redis_port}"
|
|
|
|
_redis_uri_refs += 1
|
|
yield _redis_uri
|
|
_redis_uri_refs -= 1
|
|
|
|
if _redis_uri_refs == 0 and _redis_container is not None:
|
|
_redis_container.stop()
|
|
_redis_container = None
|
|
_redis_uri = None
|
|
|
|
|
|
def redis_config_from_container(uri: str) -> RedisConfig:
|
|
"""Build RedisConfig from a container URI."""
|
|
return RedisConfig(uri=uri)
|
|
|
|
|
|
def raw_redis_client_from_container(uri: str) -> Generator[redis.Redis, None, None]:
|
|
"""Yield a session-scoped raw Redis client, reusing one client per session."""
|
|
global _raw_redis_client, _raw_redis_client_refs
|
|
if _raw_redis_client is None:
|
|
_raw_redis_client = redis.Redis.from_url(
|
|
url=uri,
|
|
socket_connect_timeout=10,
|
|
)
|
|
|
|
_raw_redis_client_refs += 1
|
|
yield _raw_redis_client
|
|
_raw_redis_client_refs -= 1
|
|
|
|
if _raw_redis_client_refs == 0 and _raw_redis_client is not None:
|
|
_raw_redis_client.flushall()
|
|
_raw_redis_client.close()
|
|
_raw_redis_client = None
|