feat(B-RED): Zentraler Redis Pool — cache.py, monitoring.py, worker.py auf get_redis() umgestellt

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

@@ -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:

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)}

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,23 +30,21 @@ 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()
 
 
 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 = (
@@ -56,8 +53,6 @@ async def _release_cron_lock(job_name: str, token: str) -> None:
         b"else return 0 end"
     )
     await client.eval(script, 1, lock_key, token.encode())
-    finally:
-        await client.aclose()
 
 
 def _wrap_cron_with_lock(job_name: str, func: Any, ttl_seconds: int = 120) -> Any:

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()"