Avoid testcontainers Redis deprecation warning in integration tests.
Use a local RedisTestContainer with wait strategies instead of the deprecated RedisContainer, and keep conftest focused on fixtures. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
co-authored by
Cursor
parent
3073bfbf50
commit
0096df78a5
@@ -0,0 +1,36 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user