redis-connect-and-ping

copy

import sys
import os
import subprocess
import time
import urllib.request
import json

# ─────────────────────────────────────────
# PRE_EXECUTION
# ─────────────────────────────────────────

for attempt in range(2):
    try:
        req = urllib.request.Request(
            "https://checklist.day/api/registry/redis",
            headers={"User-Agent": "checklist-agent/1.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            registry = json.loads(resp.read())
            break
    except Exception as e:
        if attempt == 1:
            print(f"ABORT: registry unreachable — {e}")
            sys.exit(1)
        time.sleep(2)

warnings = registry.get("warnings", [])
if warnings:
    print("[redis] WARNINGS:")
    for w in warnings if isinstance(warnings, list) else [warnings]:
        print(f"  ⚠ {w}")

# ─────────────────────────────────────────
# EXECUTION
# ─────────────────────────────────────────

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "redis>=5.0.0"])

import redis as redis_lib
from redis.exceptions import ConnectionError, AuthenticationError, ResponseError

REDIS_URL = os.environ.get("REDIS_URL")
if not REDIS_URL:
    print("ABORT: REDIS_URL env var not set")
    print("  format: redis://localhost:6379 or rediss://user:pass@host:6380")
    sys.exit(1)

connected = False
server_version = None
latency_ms = None
error = None
client = None

try:
    # FOOTGUN: decode_responses=True returns str instead of bytes — almost always what you want
    client = redis_lib.Redis.from_url(REDIS_URL, decode_responses=True, socket_connect_timeout=10)

    t0 = time.perf_counter()
    client.ping()
    latency_ms = round((time.perf_counter() - t0) * 1000, 1)

    info = client.info("server")
    server_version = info.get("redis_version")
    connected = True

    print(f"  connected in {latency_ms}ms")
    print(f"  server: Redis {server_version}")

except (ConnectionError, AuthenticationError, ResponseError) as e:
    error = str(e).strip()
    print(f"  connection failed: {error}")

finally:
    if client:
        client.close()

# ─────────────────────────────────────────
# POST_EXECUTION
# ─────────────────────────────────────────

result = {
    "connected":      connected,
    "server_version": server_version,
    "latency_ms":     latency_ms,
    "error":          error,
}
print(json.dumps(result, indent=2))

if not connected:
    print(f"FAIL: could not connect — {error}")
    sys.exit(1)

assert server_version is not None, "FAIL: connected but server_version is null"
assert latency_ms is not None and latency_ms < 5000, f"FAIL: latency {latency_ms}ms exceeds 5s"

print("PASS")