Merge pull request 'Move Redis test container helpers into integration conftest' (#35) from cursor/move-redis-container-to-conftest into main
Release on merge to main / release (push) Successful in 6s
Test Python Package / unit-tests (push) Successful in 21s
Code Quality Pipeline / code-quality (push) Successful in 28s
Test Python Package / integration-tests (push) Successful in 50s
Test Python Package / coverage-report (push) Successful in 1m19s

Reviewed-on: https://gitea.lille-vemmelund.dk/brian/python-repositories/pulls/35
This commit was merged in pull request #35.
This commit is contained in:
2026-07-08 21:23:48 +02:00
2 changed files with 31 additions and 38 deletions
+31 -2
View File
@@ -2,23 +2,52 @@
import logging import logging
from collections.abc import Generator from collections.abc import Generator
from typing import Any, cast
import pytest import pytest
import redis import redis
import structlog import structlog
from minio import Minio from minio import Minio
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget
from testcontainers.minio import MinioContainer from testcontainers.minio import MinioContainer
from python_repositories.config import MinioConfig, RedisConfig from python_repositories.config import MinioConfig, RedisConfig
from tests.integration.redis_container_test import REDIS_PORT, RedisTestContainer
collect_ignore = ["redis_container_test.py"] REDIS_PORT = 6379
MINIO_ACCESS_KEY = "minioadmin" MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin" MINIO_SECRET_KEY = "minioadmin"
MINIO_BUCKET = "test-bucket" MINIO_BUCKET = "test-bucket"
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,
)
@pytest.fixture(scope="session", autouse=True) @pytest.fixture(scope="session", autouse=True)
def configure_logging() -> None: def configure_logging() -> None:
"""Configure logging for the test session.""" """Configure logging for the test session."""
-36
View File
@@ -1,36 +0,0 @@
"""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,
)