harness_zbt/backend/tests/test_redis_keys.py

import pytest

from zbt.infrastructure.redis.keys import RedisKeyBuilder


def test_redis_key_builder_uses_stage_prefix_and_rejects_unsafe_segments() -> None:
    keys = RedisKeyBuilder("ins:s3:")

    assert keys.auth_otp("mobile-sha256") == "ins:s3:auth:otp:mobile-sha256"
    assert keys.auth_session("session-01") == "ins:s3:auth:session:session-01"
    assert keys.agent_run_events("run-01") == "ins:s3:agent:run:run-01:events"
    assert keys.payment_lock("payment-01") == "ins:s3:lock:payment:payment-01"
    assert keys.worker_lock("refund") == "ins:s3:worker:lock:refund"

    with pytest.raises(ValueError, match="Redis Key"):
        keys.auth_session("other:namespace")