From e9164979b5bf631ed7d2b0660905fc5c550b2cb8 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 13 Aug 2026 16:25:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(B-RED):=20Zentraler=20Redis=20Pool=20?= =?UTF-8?q?=E2=80=94=20cache.py,=20monitoring.py,=20worker.py=20auf=20get?= =?UTF-8?q?=5Fredis()=20umgestellt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B-RED: 5 direkte aioredis.from_url() Konstruktoren auf get_redis() umgestellt - cache.py: get_cache() delegiert auf get_redis(), _cache_redis Singleton entfernt - monitoring.py: check_redis() und check_worker() nutzen get_redis() - worker.py: _acquire_cron_lock() und _release_cron_lock() nutzen get_redis() - 0 verbleibende direkte aioredis.from_url() außerhalb auth.py B-RED-TEST: 8 Tests in test_redis_pool.py — alle grün - Singleton, get_cache delegation, SET/GET, parallel, reset, no-direct-from_url checks --- app/core/cache.py | 11 ++--- app/core/monitoring.py | 16 ++---- app/core/worker.py | 37 ++++++-------- tests/test_redis_pool.py | 102 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 41 deletions(-) create mode 100644 tests/test_redis_pool.py diff --git a/app/core/cache.py b/app/core/cache.py index 5652e07..32fd2df 100644 --- a/app/core/cache.py +++ b/app/core/cache.py @@ -7,17 +7,12 @@ from typing import Any import redis.asyncio as aioredis -from app.config import get_settings - -_cache_redis: aioredis.Redis | None = None +from app.core.auth import get_redis def get_cache() -> aioredis.Redis: - """Get or create the cache Redis client.""" - global _cache_redis - if _cache_redis is None: - _cache_redis = aioredis.from_url(get_settings().redis_url, decode_responses=True) - return _cache_redis + """Get the cache Redis client (delegates to the global ``get_redis()`` singleton).""" + return get_redis() async def cache_get(key: str) -> Any | None: diff --git a/app/core/monitoring.py b/app/core/monitoring.py index a6850a2..538dc78 100644 --- a/app/core/monitoring.py +++ b/app/core/monitoring.py @@ -149,14 +149,10 @@ async def check_database() -> dict[str, Any]: async def check_redis() -> dict[str, Any]: """Check Redis connectivity (async).""" try: - import redis.asyncio as aioredis + from app.core.auth import get_redis - from app.config import get_settings - - settings = get_settings() - r = aioredis.from_url(settings.redis_url, decode_responses=True) + r = get_redis() pong = await r.ping() - await r.aclose() if pong: return {"status": "up", "latency_ms": 0} return {"status": "down", "error": "Redis returned False for PING"} @@ -185,15 +181,11 @@ async def check_worker() -> dict[str, Any]: In test/dev mode this checks if Redis is available for the worker queue. """ try: - import redis.asyncio as aioredis + from app.core.auth import get_redis - from app.config import get_settings - - settings = get_settings() - r = aioredis.from_url(settings.redis_url, decode_responses=True) + r = get_redis() # Check if arq queue key exists queue_length = await r.zcard("arq:queue") - await r.aclose() return {"status": "up", "queue_length": queue_length} except Exception as e: return {"status": "down", "error": str(e)} diff --git a/app/core/worker.py b/app/core/worker.py index 8d42a86..9bfe6eb 100644 --- a/app/core/worker.py +++ b/app/core/worker.py @@ -20,7 +20,6 @@ logger = logging.getLogger(__name__) # on every replica. We use a short-lived Redis SET NX lock per cron call # so only one replica actually executes the job. -import redis.asyncio as aioredis # noqa: E402 import uuid # noqa: E402 @@ -31,33 +30,29 @@ async def _acquire_cron_lock(job_name: str, ttl_seconds: int = 120) -> str | Non replica already holds the lock. The lock auto-expires after *ttl_seconds* to avoid deadlocks if a worker crashes mid-job. """ - settings = get_settings() - client = aioredis.from_url(settings.redis_url) + from app.core.auth import get_redis + + client = get_redis() token = str(uuid.uuid4()) lock_key = f"leocrm:cron_lock:{job_name}" - try: - acquired = await client.set(lock_key, token, nx=True, ex=ttl_seconds) - return token if acquired else None - finally: - await client.aclose() + acquired = await client.set(lock_key, token, nx=True, ex=ttl_seconds) + return token if acquired else None async def _release_cron_lock(job_name: str, token: str) -> None: """Release a previously acquired cron lock using a safe compare-and-delete.""" - settings = get_settings() - client = aioredis.from_url(settings.redis_url) + from app.core.auth import get_redis + + client = get_redis() lock_key = f"leocrm:cron_lock:{job_name}" - try: - # Lua script ensures we only delete if the token matches (avoid - # releasing a lock that was already expired and re-acquired). - script = ( - b"if redis.call('get', KEYS[1]) == ARGV[1] " - b"then return redis.call('del', KEYS[1]) " - b"else return 0 end" - ) - await client.eval(script, 1, lock_key, token.encode()) - finally: - await client.aclose() + # Lua script ensures we only delete if the token matches (avoid + # releasing a lock that was already expired and re-acquired). + script = ( + b"if redis.call('get', KEYS[1]) == ARGV[1] " + b"then return redis.call('del', KEYS[1]) " + b"else return 0 end" + ) + await client.eval(script, 1, lock_key, token.encode()) def _wrap_cron_with_lock(job_name: str, func: Any, ttl_seconds: int = 120) -> Any: diff --git a/tests/test_redis_pool.py b/tests/test_redis_pool.py new file mode 100644 index 0000000..335e9a2 --- /dev/null +++ b/tests/test_redis_pool.py @@ -0,0 +1,102 @@ +"""Tests for central Redis pool consolidation (B-RED-TEST). + +Verifies that get_redis() returns a singleton, get_cache() delegates to it, +the connection works, parallel calls are safe, and reset works. +""" + +from __future__ import annotations + +import asyncio + +import pytest +import redis.asyncio as aioredis + +from app.core import auth +from app.core.auth import close_redis, get_redis +from app.core.cache import get_cache + + +@pytest.fixture(autouse=True) +def _reset_redis_singleton(): + """Ensure a clean Redis singleton before and after each test.""" + # Reset before test + auth._redis_client = None + yield + # Reset after test + auth._redis_client = None + + +@pytest.mark.asyncio +async def test_get_redis_returns_same_instance(): + """get_redis() must return the same instance on repeated calls.""" + r1 = get_redis() + r2 = get_redis() + assert r1 is r2, "get_redis() returned different instances" + + +@pytest.mark.asyncio +async def test_get_cache_returns_same_instance_as_get_redis(): + """get_cache() must delegate to get_redis() and return the same instance.""" + r = get_redis() + c = get_cache() + assert c is r, "get_cache() did not return the same instance as get_redis()" + + +@pytest.mark.asyncio +async def test_get_redis_connection_works(): + """The Redis connection from get_redis() must support SET/GET.""" + r = get_redis() + await r.set("test:pool:key", "hello") + val = await r.get("test:pool:key") + assert val == "hello" + await r.delete("test:pool:key") + + +@pytest.mark.asyncio +async def test_parallel_get_redis_returns_same_instance(): + """Multiple parallel get_redis() calls must return the same instance.""" + results = await asyncio.gather(*[asyncio.to_thread(get_redis) for _ in range(10)]) + first = results[0] + assert all(r is first for r in results), "Parallel get_redis() returned different instances" + + +@pytest.mark.asyncio +async def test_close_redis_resets_instance(): + """close_redis() must reset the singleton so the next get_redis() creates a new one.""" + r1 = get_redis() + await close_redis() + r2 = get_redis() + assert r1 is not r2, "close_redis() did not reset the singleton" + + +@pytest.mark.asyncio +async def test_no_direct_aioredis_from_url_in_cache(): + """cache.py must not use aioredis.from_url() directly (uses get_redis() instead).""" + import inspect + + from app.core import cache + + source = inspect.getsource(cache) + assert "from_url" not in source, "cache.py still contains aioredis.from_url()" + + +@pytest.mark.asyncio +async def test_no_direct_aioredis_from_url_in_monitoring(): + """monitoring.py must not use aioredis.from_url() directly.""" + import inspect + + from app.core import monitoring + + source = inspect.getsource(monitoring) + assert "from_url" not in source, "monitoring.py still contains aioredis.from_url()" + + +@pytest.mark.asyncio +async def test_no_direct_aioredis_from_url_in_worker(): + """worker.py must not use aioredis.from_url() directly.""" + import inspect + + from app.core import worker + + source = inspect.getsource(worker) + assert "from_url" not in source, "worker.py still contains aioredis.from_url()"