Brian Bjarke JensenandCursor
0096df78a5
Use a local RedisTestContainer with wait strategies instead of the deprecated RedisContainer, and keep conftest focused on fixtures. Co-authored-by: Cursor <[email protected]>
37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
"""Redis test container without testcontainers' deprecated wait decorator."""
|
|
|
|
from typing import Any, cast
|
|
|
|
import redis
|
|
from testcontainers.core.container import DockerContainer
|
|
from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget
|
|
|
|
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,
|
|
)
|