redis-set-get-with-ttl

code_execution · unverified · null · json · download .py

Redis connection URL e.g. "redis://localhost:6379"

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

REDIS_URL = os.environ.get("REDIS_URL")
if not REDIS_URL:
    print("ABORT: REDIS_URL env var not set")
    sys.exit(1)

KEY    = "checklist:ttl:test"
VALUE  = "hello-checklist"
TTL    = 60  # seconds

client = redis_lib.Redis.from_url(REDIS_URL, decode_responses=True, socket_connect_timeout=10)

try:
    # Cleanup
    client.delete(KEY)

    # 1. SET with TTL
    # FOOTGUN: use SET key value EX seconds (atomic) — not separate SET then EXPIRE
    # FOOTGUN: SETEX is deprecated since Redis 2.6 — use SET with EX parameter
    set_ok = client.set(KEY, VALUE, ex=TTL)
    print(f"  SET {KEY} = '{VALUE}' EX {TTL}s → {set_ok}")

    # 2. GET
    retrieved = client.get(KEY)
    get_ok = retrieved == VALUE
    print(f"  GET {KEY} → '{retrieved}' (match={get_ok})")

    # 3. Verify TTL
    # FOOTGUN: TTL returns -1 if key has no expiry, -2 if key doesn't exist
    ttl_verified = client.ttl(KEY)
    expiry_ok = 0 < ttl_verified <= TTL
    print(f"  TTL {KEY} → {ttl_verified}s (expected <= {TTL}s, expiry_ok={expiry_ok})")

    # Cleanup
    client.delete(KEY)

finally:
    client.close()

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

assert set_ok, "FAIL: SET returned falsy"
assert get_ok, f"FAIL: GET returned '{retrieved}', expected '{VALUE}'"
assert expiry_ok, f"FAIL: TTL={ttl_verified} out of expected range (0, {TTL}]"

result = {
    "set_ok":       bool(set_ok),
    "get_ok":       get_ok,
    "ttl_set":      TTL,
    "ttl_verified": ttl_verified,
    "expiry_ok":    expiry_ok,
}
print(json.dumps(result, indent=2))
print("PASS")