Agentswarm/scripts/test-redis-connection-config.py

#!/usr/bin/env python3
"""Unit test for RedisClient connection config (TLS / password / REDIS_URL).

Validates that env config maps to the right redis-py connection kwargs WITHOUT
opening a connection — so it runs anywhere, no live Redis needed. Covers:
  - default plaintext in-cluster behaviour (backward compat)
  - discrete REDIS_PASSWORD / REDIS_SSL
  - REDIS_URL precedence + rediss:// (TLS)
  - logs never leak the URL/password (credential-safe target description)

Run: python scripts/test-redis-connection-config.py
"""
import importlib
import os
import sys

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if REPO_ROOT not in sys.path:
    sys.path.insert(0, REPO_ROOT)

REDIS_ENV_VARS = (
    "REDIS_URL", "REDIS_HOST", "REDIS_PORT", "REDIS_DB",
    "REDIS_PASSWORD", "REDIS_SSL", "REDIS_FAKE", "ALLOW_MEMORY_STORE",
)


def _fresh_client(env):
    """Reset the relevant env, apply `env`, and return a freshly built RedisClient."""
    for k in REDIS_ENV_VARS:
        os.environ.pop(k, None)
    os.environ.update(env)
    import orchestrator.redis_client as rc
    importlib.reload(rc)
    return rc.RedisClient(), rc


def _conn_kwargs(client_obj):
    """Pull the connection kwargs redis-py resolved for the built client."""
    real = client_obj._make_real_client()
    return real.connection_pool.connection_kwargs, real


def test_default_plaintext():
    c, _ = _fresh_client({})
    kw, real = _conn_kwargs(c)
    assert kw.get("host") == "redis-service", kw
    assert kw.get("port") == 6379, kw
    assert "password" not in kw or kw["password"] is None, kw
    # No TLS: connection class must not be the SSL one.
    assert "SSL" not in real.connection_pool.connection_class.__name__, kw
    assert "TLS=False" in c._target_desc()
    print("ok: default plaintext (backward compatible)")


def test_discrete_password_and_ssl():
    c, _ = _fresh_client({
        "REDIS_HOST": "heicode-rd.redisenterprise.cache.azure.net",
        "REDIS_PORT": "10000",
        "REDIS_PASSWORD": "s3cr3t-from-secret",
        "REDIS_SSL": "1",
    })
    kw, real = _conn_kwargs(c)
    assert kw.get("host") == "heicode-rd.redisenterprise.cache.azure.net", kw
    assert kw.get("port") == 10000, kw
    assert kw.get("password") == "s3cr3t-from-secret", kw
    assert "SSL" in real.connection_pool.connection_class.__name__, "expected TLS connection class"
    # target description must NOT leak the password
    assert "s3cr3t" not in c._target_desc(), c._target_desc()
    assert "TLS=True" in c._target_desc()
    print("ok: discrete REDIS_PASSWORD + REDIS_SSL → authenticated TLS")


def test_url_precedence_and_tls():
    c, _ = _fresh_client({
        "REDIS_URL": "rediss://:url-password@heicode-rd.example.net:10000/0",
        # discrete vars set too — URL must win
        "REDIS_HOST": "redis-service",
        "REDIS_SSL": "",
    })
    kw, real = _conn_kwargs(c)
    assert kw.get("host") == "heicode-rd.example.net", kw
    assert kw.get("port") == 10000, kw
    assert kw.get("password") == "url-password", kw
    assert "SSL" in real.connection_pool.connection_class.__name__, "rediss:// must enable TLS"
    desc = c._target_desc()
    assert "url-password" not in desc and "heicode-rd.example.net" not in desc, desc
    assert "TLS=True" in desc
    print("ok: REDIS_URL precedence + rediss:// TLS, credentials not logged")


def test_cluster_mode():
    # OSS Cluster endpoint (e.g. Azure Redis Enterprise clusteringPolicy=OSSCluster):
    # must build a cluster-aware client so keys()/cross-slot routing is correct.
    c, _ = _fresh_client({
        "REDIS_CLUSTER": "1",
        "REDIS_HOST": "heicode-rd.southeastasia.redis.azure.net",
        "REDIS_PORT": "10000",
        "REDIS_PASSWORD": "key-from-secret",
    })
    real = c._make_real_client()  # constructs lazily; does NOT connect
    assert type(real).__name__ == "RedisCluster", type(real).__name__
    assert "cluster" in c._target_desc(), c._target_desc()
    assert "key-from-secret" not in c._target_desc(), c._target_desc()
    print("ok: REDIS_CLUSTER → RedisCluster client (OSS Cluster safe)")


def main():
    try:
        import redis  # noqa: F401
    except Exception:
        print("SKIP: redis-py not installed; connection-config test requires it")
        return 0
    test_default_plaintext()
    test_discrete_password_and_ssl()
    test_url_precedence_and_tls()
    test_cluster_mode()
    print("\nALL PASS: redis connection config")
    return 0


if __name__ == "__main__":
    sys.exit(main())