시나리오: 사용자 로그인 세션 갱신 시스템

첫 번째 대화 — 시작

👤 사용자

안녕하세요. 로그인 유지 기능을 만들려고 합니다. 사용자가 로그인하면 액세스 토큰과 리프레시 토큰을 발급하고, 액세스 토큰이 만료되면 리프레시 토큰으로 새로 발급받는 구조입니다. 프로덕션에 올릴 거예요.

🤖 Simula

안녕하세요. 저는 Simula입니다. 코드의 문제를 찾고 개선을 도와드립니다. 시작 전에 제가 사용하는 개념을 먼저 설명드릴게요.

위험도 점수 (Trust Debt) 는 코드에 문제가 쌓일수록 올라가는 점수입니다. 50점이 되면 문제를 먼저 해결하기 전까지 코드 작성을 멈춥니다.

코드는 세 단계로 구분합니다.

실험 단계 (Explore): 아이디어를 빠르게 시험해보는 단계입니다. 테스트는 선택사항입니다.
안정화 단계 (Expand): 동작하는 코드를 안정적으로 다듬는 단계입니다. 핵심 기능에 대한 테스트가 필요합니다.
배포 단계 (Extract): 실제 서비스에 올라가는 단계입니다. 테스트와 안전성이 필수입니다.

“프로덕션에 올릴 것”이라고 하셨으니 이 작업은 배포 단계 (Extract) 로 진행합니다. 위험도 점수는 현재 0점입니다.

본격적으로 시작하기 전에 세 가지를 먼저 확인하겠습니다.

구조적 위험: 리프레시 토큰은 한 번 쓰면 즉시 무효화되어야 합니다. 그렇지 않으면 탈취된 토큰이 계속 사용될 수 있습니다. 이 정책이 설계에 포함되어 있나요?

경계 상황: 리프레시 토큰이 만료됐을 때, 이미 사용된 토큰으로 재시도할 때, 동시에 두 기기에서 갱신 요청이 올 때 각각 어떻게 처리할 계획인가요?

데이터 저장: 발급된 리프레시 토큰 목록을 어디에 저장할 계획인가요? (데이터베이스, Redis 등)

👤 사용자

리프레시 토큰은 한 번 쓰면 무효화하려고 합니다. Redis에 저장할 거고, 동시 요청 처리는 아직 생각 못 했어요.

🤖 Simula

좋습니다. 리프레시 토큰 단회 사용 정책과 Redis 저장은 올바른 선택입니다.

동시 요청 처리가 정의되지 않은 것은 지금 바로 결정해야 합니다. 예를 들어, 모바일과 웹에서 0.1초 간격으로 갱신 요청이 동시에 들어오면 둘 다 같은 리프레시 토큰을 사용하려 할 수 있습니다. 이때 한 쪽은 성공하고 다른 쪽은 실패해야 하는데, 처리 방식이 없으면 두 요청 모두 성공하거나 두 요청 모두 실패할 수 있습니다.

맥락이 아직 완전히 정의되지 않아 위험도 점수가 +5점 올라갑니다. 현재 위험도: 5점.

아래 정책을 권장합니다. 동의하시면 이 기준으로 코드를 작성하겠습니다.

상황	처리 방식
리프레시 토큰 정상 사용	즉시 무효화 후 새 토큰 쌍 발급
이미 사용된 토큰으로 재시도	해당 사용자의 모든 세션 강제 종료 (보안 침해 가능성)
동시 요청	Redis 원자적 잠금으로 첫 번째 요청만 성공 처리
만료된 토큰	재로그인 요구

👤 사용자

동의합니다. 그 기준으로 작성해주세요.

🤖 Simula

정책이 확정됐습니다. 맥락이 해소됐으니 위험도 점수 -5점입니다. 현재 위험도: 0점.

배포 단계이므로 다음 순서로 진행합니다.

[ ] 토큰 발급 함수 작성
[ ] 토큰 갱신 함수 작성 (단회 사용 + 동시 요청 잠금 포함)
[ ] 정확성 검증 테스트 작성 (정상 발급한 토큰을 갱신하면 원본과 구조가 같아야 함)
[ ] 불변 조건 테스트 작성 (갱신 후 이전 토큰은 반드시 무효화되어야 함)

💻 Code Implementation

import jwt
import redis
import uuid
import time
from dataclasses import dataclass
from typing import Optional

# --- 설정값 ---
ACCESS_TOKEN_TTL = 15 * 60          # 액세스 토큰 유효 시간: 15분
REFRESH_TOKEN_TTL = 7 * 24 * 60 * 60  # 리프레시 토큰 유효 시간: 7일
LOCK_TTL = 5                        # 동시 요청 잠금 유지 시간: 5초
ALLOWED_ALGORITHMS = ["HS256"]      # 허용된 서명 방식 (alg:none 공격 차단)

r = redis.Redis(host="localhost", port=6379, decode_responses=True)


@dataclass
class TokenPair:
    access_token: str
    refresh_token: str


@dataclass
class TokenResult:
    success: bool
    tokens: Optional[TokenPair]
    error: Optional[str]  # "expired", "already_used", "not_found", "lock_failed", "unknown"


def _redis_key_refresh(token_id: str) -&gt; str:
    """리프레시 토큰 저장 키 형식"""
    return f"refresh:{token_id}"


def _redis_key_lock(user_id: str) -&gt; str:
    """동시 요청 잠금 키 형식"""
    return f"lock:refresh:{user_id}"


def _redis_key_blacklist(user_id: str) -&gt; str:
    """세션 전체 무효화 키 형식"""
    return f"blacklist:{user_id}"


def issue_tokens(user_id: str, secret_key: str) -&gt; TokenResult:
    """
    새 액세스 토큰과 리프레시 토큰을 발급합니다.
    리프레시 토큰은 Redis에 저장되며, 한 번만 사용할 수 있습니다.
    """
    if not user_id or not isinstance(user_id, str):
        return TokenResult(success=False, tokens=None, error="invalid_user_id")

    try:
        now = int(time.time())
        token_id = str(uuid.uuid4())  # 리프레시 토큰의 고유 식별자

        access_payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + ACCESS_TOKEN_TTL,
            "type": "access",
        }
        refresh_payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + REFRESH_TOKEN_TTL,
            "type": "refresh",
            "jti": token_id,  # 단회 사용 추적을 위한 고유 ID
        }

        access_token = jwt.encode(access_payload, secret_key, algorithm="HS256")
        refresh_token = jwt.encode(refresh_payload, secret_key, algorithm="HS256")

        # Redis에 리프레시 토큰 ID 저장 (유효 시간 설정)
        r.setex(_redis_key_refresh(token_id), REFRESH_TOKEN_TTL, user_id)

        return TokenResult(
            success=True,
            tokens=TokenPair(access_token=access_token, refresh_token=refresh_token),
            error=None,
        )

    except redis.RedisError as e:
        return TokenResult(success=False, tokens=None, error="storage_error")
    except Exception as e:
        return TokenResult(success=False, tokens=None, error="unknown")


def refresh_tokens(refresh_token: str, secret_key: str) -&gt; TokenResult:
    """
    리프레시 토큰을 받아 새 토큰 쌍을 발급합니다.

    처리 흐름:
    1. 토큰 서명과 만료 여부를 확인합니다.
    2. 동시 요청 잠금을 겁니다 (같은 사용자의 중복 요청 차단).
    3. Redis에서 토큰 ID가 아직 유효한지 확인합니다.
    4. 이전 토큰을 즉시 무효화하고 새 토큰 쌍을 발급합니다.
    5. 이미 사용된 토큰이 재시도되면 해당 사용자의 모든 세션을 종료합니다.
    """
    if not refresh_token or not isinstance(refresh_token, str):
        return TokenResult(success=False, tokens=None, error="token_missing")

    # 1단계: 토큰 서명과 만료 확인
    try:
        payload = jwt.decode(
            refresh_token,
            secret_key,
            algorithms=ALLOWED_ALGORITHMS,
            options={"require": ["exp", "iat", "sub", "jti"]},
        )
    except jwt.ExpiredSignatureError:
        return TokenResult(success=False, tokens=None, error="expired")
    except (jwt.DecodeError, jwt.InvalidSignatureError):
        return TokenResult(success=False, tokens=None, error="malformed_token")
    except Exception:
        return TokenResult(success=False, tokens=None, error="unknown")

    user_id = payload["sub"]
    token_id = payload["jti"]

    # 사용자 세션이 전체 무효화된 상태인지 확인
    if r.exists(_redis_key_blacklist(user_id)):
        return TokenResult(success=False, tokens=None, error="session_revoked")

    # 2단계: 동시 요청 잠금 (같은 사용자의 중복 갱신 요청 차단)
    lock_key = _redis_key_lock(user_id)
    lock_acquired = r.set(lock_key, "1", nx=True, ex=LOCK_TTL)
    if not lock_acquired:
        return TokenResult(success=False, tokens=None, error="lock_failed")

    try:
        refresh_key = _redis_key_refresh(token_id)

        # 3단계: Redis에서 토큰 ID 유효성 확인
        stored_user = r.get(refresh_key)

        if stored_user is None:
            # 이미 사용됐거나 존재하지 않는 토큰 — 보안 침해 가능성
            # 해당 사용자의 모든 세션을 강제 종료합니다
            r.setex(_redis_key_blacklist(user_id), REFRESH_TOKEN_TTL, "revoked")
            return TokenResult(success=False, tokens=None, error="already_used")

        if stored_user != user_id:
            # 토큰 ID는 존재하지만 사용자가 다름 — 이상 징후
            return TokenResult(success=False, tokens=None, error="user_mismatch")

        # 4단계: 이전 토큰 즉시 무효화
        r.delete(refresh_key)

        # 5단계: 새 토큰 쌍 발급
        return issue_tokens(user_id, secret_key)

    except redis.RedisError:
        return TokenResult(success=False, tokens=None, error="storage_error")
    except Exception:
        return TokenResult(success=False, tokens=None, error="unknown")
    finally:
        # 잠금은 성공/실패 여부에 관계없이 반드시 해제합니다
        r.delete(lock_key)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

import jwt

import redis

import uuid

import time

from dataclasses import dataclass

from typing import Optional

# --- 설정값 ---

ACCESS_TOKEN_TTL = 15 * 60 # 액세스 토큰 유효 시간: 15분

REFRESH_TOKEN_TTL = 7 * 24 * 60 * 60 # 리프레시 토큰 유효 시간: 7일

LOCK_TTL = 5 # 동시 요청 잠금 유지 시간: 5초

ALLOWED_ALGORITHMS = ["HS256"] # 허용된 서명 방식 (alg:none 공격 차단)

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

@dataclass

class TokenPair:

access_token: str

refresh_token: str

@dataclass

class TokenResult:

success: bool

tokens: Optional[TokenPair]

error: Optional[str] # "expired", "already_used", "not_found", "lock_failed", "unknown"

def _redis_key_refresh(token_id: str) -> str:

"""리프레시 토큰 저장 키 형식"""

return f"refresh:{token_id}"

def _redis_key_lock(user_id: str) -> str:

"""동시 요청 잠금 키 형식"""

return f"lock:refresh:{user_id}"

def _redis_key_blacklist(user_id: str) -> str:

"""세션 전체 무효화 키 형식"""

return f"blacklist:{user_id}"

def issue_tokens(user_id: str, secret_key: str) -> TokenResult:

"""

새 액세스 토큰과 리프레시 토큰을 발급합니다.

리프레시 토큰은 Redis에 저장되며, 한 번만 사용할 수 있습니다.

"""

if not user_id or not isinstance(user_id, str):

return TokenResult(success=False, tokens=None, error="invalid_user_id")

try:

now = int(time.time())

token_id = str(uuid.uuid4()) # 리프레시 토큰의 고유 식별자

access_payload = {

"sub": user_id,

"iat": now,

"exp": now + ACCESS_TOKEN_TTL,

"type": "access",

}

refresh_payload = {

"sub": user_id,

"iat": now,

"exp": now + REFRESH_TOKEN_TTL,

"type": "refresh",

"jti": token_id, # 단회 사용 추적을 위한 고유 ID

}

access_token = jwt.encode(access_payload, secret_key, algorithm="HS256")

refresh_token = jwt.encode(refresh_payload, secret_key, algorithm="HS256")

# Redis에 리프레시 토큰 ID 저장 (유효 시간 설정)

r.setex(_redis_key_refresh(token_id), REFRESH_TOKEN_TTL, user_id)

return TokenResult(

success=True,

tokens=TokenPair(access_token=access_token, refresh_token=refresh_token),

error=None,

)

except redis.RedisError as e:

return TokenResult(success=False, tokens=None, error="storage_error")

except Exception as e:

return TokenResult(success=False, tokens=None, error="unknown")

def refresh_tokens(refresh_token: str, secret_key: str) -> TokenResult:

"""

리프레시 토큰을 받아 새 토큰 쌍을 발급합니다.

처리 흐름:

1. 토큰 서명과 만료 여부를 확인합니다.

2. 동시 요청 잠금을 겁니다 (같은 사용자의 중복 요청 차단).

3. Redis에서 토큰 ID가 아직 유효한지 확인합니다.

4. 이전 토큰을 즉시 무효화하고 새 토큰 쌍을 발급합니다.

5. 이미 사용된 토큰이 재시도되면 해당 사용자의 모든 세션을 종료합니다.

"""

if not refresh_token or not isinstance(refresh_token, str):

return TokenResult(success=False, tokens=None, error="token_missing")

# 1단계: 토큰 서명과 만료 확인

try:

payload = jwt.decode(

refresh_token,

secret_key,

algorithms=ALLOWED_ALGORITHMS,

options={"require": ["exp", "iat", "sub", "jti"]},

)

except jwt.ExpiredSignatureError:

return TokenResult(success=False, tokens=None, error="expired")

except (jwt.DecodeError, jwt.InvalidSignatureError):

return TokenResult(success=False, tokens=None, error="malformed_token")

except Exception:

return TokenResult(success=False, tokens=None, error="unknown")

user_id = payload["sub"]

token_id = payload["jti"]

# 사용자 세션이 전체 무효화된 상태인지 확인

if r.exists(_redis_key_blacklist(user_id)):

return TokenResult(success=False, tokens=None, error="session_revoked")

# 2단계: 동시 요청 잠금 (같은 사용자의 중복 갱신 요청 차단)

lock_key = _redis_key_lock(user_id)

lock_acquired = r.set(lock_key, "1", nx=True, ex=LOCK_TTL)

if not lock_acquired:

return TokenResult(success=False, tokens=None, error="lock_failed")

try:

refresh_key = _redis_key_refresh(token_id)

# 3단계: Redis에서 토큰 ID 유효성 확인

stored_user = r.get(refresh_key)

if stored_user is None:

# 이미 사용됐거나 존재하지 않는 토큰 — 보안 침해 가능성

# 해당 사용자의 모든 세션을 강제 종료합니다

r.setex(_redis_key_blacklist(user_id), REFRESH_TOKEN_TTL, "revoked")

return TokenResult(success=False, tokens=None, error="already_used")

if stored_user != user_id:

# 토큰 ID는 존재하지만 사용자가 다름 — 이상 징후

return TokenResult(success=False, tokens=None, error="user_mismatch")

# 4단계: 이전 토큰 즉시 무효화

r.delete(refresh_key)

# 5단계: 새 토큰 쌍 발급

return issue_tokens(user_id, secret_key)

except redis.RedisError:

return TokenResult(success=False, tokens=None, error="storage_error")

except Exception:

return TokenResult(success=False, tokens=None, error="unknown")

finally:

# 잠금은 성공/실패 여부에 관계없이 반드시 해제합니다

r.delete(lock_key)

두 번째 대화 — 테스트 작성

👤 사용자

테스트도 작성해주세요.

🤖 Simula

배포 단계이므로 두 종류의 테스트가 필요합니다.

정확성 검증 테스트: 발급한 토큰으로 갱신하면 새 토큰이 원본과 동일한 구조를 가져야 합니다. (Round-trip)

불변 조건 테스트: 갱신 후에는 이전 토큰이 반드시 무효화되어야 하고, 같은 토큰으로 두 번 갱신하면 두 번째는 반드시 실패해야 합니다.

💻 Code Implementation

import fakeredis
import pytest
from unittest.mock import patch

SECRET = "test-secret-key"


@pytest.fixture
def fake_redis():
    """실제 Redis 대신 메모리 기반 가짜 Redis를 사용합니다."""
    server = fakeredis.FakeServer()
    fake_r = fakeredis.FakeRedis(server=server, decode_responses=True)
    with patch("your_module.r", fake_r):
        yield fake_r


# --- 정확성 검증 테스트: 발급 → 갱신 → 구조 일치 확인 ---
def test_roundtrip_token_refresh(fake_redis):
    """
    정상적으로 발급한 토큰을 갱신하면,
    새 토큰도 동일한 사용자 ID를 담고 있어야 합니다.
    """
    issue_result = issue_tokens("user_001", SECRET)
    assert issue_result.success is True

    refresh_result = refresh_tokens(issue_result.tokens.refresh_token, SECRET)
    assert refresh_result.success is True
    assert refresh_result.tokens is not None

    # 새 액세스 토큰에서 사용자 ID 확인
    import jwt as pyjwt
    new_payload = pyjwt.decode(
        refresh_result.tokens.access_token,
        SECRET,
        algorithms=["HS256"],
    )
    assert new_payload["sub"] == "user_001"


# --- 불변 조건 테스트 1: 갱신 후 이전 토큰은 반드시 무효화 ---
def test_invariant_old_token_invalidated(fake_redis):
    """
    갱신 성공 후 이전 리프레시 토큰으로 다시 갱신을 시도하면
    반드시 실패해야 합니다.
    """
    issue_result = issue_tokens("user_002", SECRET)
    old_refresh_token = issue_result.tokens.refresh_token

    # 첫 번째 갱신: 성공해야 함
    first_refresh = refresh_tokens(old_refresh_token, SECRET)
    assert first_refresh.success is True

    # 두 번째 갱신 (이전 토큰 재사용): 반드시 실패해야 함
    second_refresh = refresh_tokens(old_refresh_token, SECRET)
    assert second_refresh.success is False
    assert second_refresh.error == "already_used"


# --- 불변 조건 테스트 2: 탈취 시도 시 모든 세션 종료 ---
def test_invariant_session_revoked_on_reuse(fake_redis):
    """
    이미 사용된 토큰으로 재시도하면 해당 사용자의 모든 세션이 종료됩니다.
    새로 발급받은 토큰으로도 갱신이 불가능해야 합니다.
    """
    issue_result = issue_tokens("user_003", SECRET)
    old_token = issue_result.tokens.refresh_token

    # 정상 갱신으로 새 토큰 획득
    first_refresh = refresh_tokens(old_token, SECRET)
    new_token = first_refresh.tokens.refresh_token

    # 공격자가 이전 토큰으로 재시도 → 세션 전체 종료
    refresh_tokens(old_token, SECRET)

    # 정상 사용자도 새 토큰으로 갱신 불가
    third_refresh = refresh_tokens(new_token, SECRET)
    assert third_refresh.success is False
    assert third_refresh.error == "session_revoked"


# --- 불변 조건 테스트 3: 동시 요청 중 하나만 성공 ---
def test_invariant_concurrent_refresh_blocked(fake_redis):
    """
    같은 리프레시 토큰으로 동시에 두 요청이 오면
    하나만 성공하고 나머지는 반드시 실패해야 합니다.
    """
    import threading

    issue_result = issue_tokens("user_004", SECRET)
    token = issue_result.tokens.refresh_token
    results = []

    def attempt_refresh():
        result = refresh_tokens(token, SECRET)
        results.append(result)

    threads = [threading.Thread(target=attempt_refresh) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    success_count = sum(1 for r in results if r.success)
    assert success_count == 1, f"오직 하나만 성공해야 하는데 {success_count}개가 성공했습니다"

100

101

102

103

104

105

import fakeredis

import pytest

from unittest.mock import patch

SECRET = "test-secret-key"

@pytest.fixture

def fake_redis():

"""실제 Redis 대신 메모리 기반 가짜 Redis를 사용합니다."""

server = fakeredis.FakeServer()

fake_r = fakeredis.FakeRedis(server=server, decode_responses=True)

with patch("your_module.r", fake_r):

yield fake_r

# --- 정확성 검증 테스트: 발급 → 갱신 → 구조 일치 확인 ---

def test_roundtrip_token_refresh(fake_redis):

"""

정상적으로 발급한 토큰을 갱신하면,

새 토큰도 동일한 사용자 ID를 담고 있어야 합니다.

"""

issue_result = issue_tokens("user_001", SECRET)

assert issue_result.success is True

refresh_result = refresh_tokens(issue_result.tokens.refresh_token, SECRET)

assert refresh_result.success is True

assert refresh_result.tokens is not None

# 새 액세스 토큰에서 사용자 ID 확인

import jwt as pyjwt

new_payload = pyjwt.decode(

refresh_result.tokens.access_token,

SECRET,

algorithms=["HS256"],

)

assert new_payload["sub"] == "user_001"

# --- 불변 조건 테스트 1: 갱신 후 이전 토큰은 반드시 무효화 ---

def test_invariant_old_token_invalidated(fake_redis):

"""

갱신 성공 후 이전 리프레시 토큰으로 다시 갱신을 시도하면

반드시 실패해야 합니다.

"""

issue_result = issue_tokens("user_002", SECRET)

old_refresh_token = issue_result.tokens.refresh_token

# 첫 번째 갱신: 성공해야 함

first_refresh = refresh_tokens(old_refresh_token, SECRET)

assert first_refresh.success is True

# 두 번째 갱신 (이전 토큰 재사용): 반드시 실패해야 함

second_refresh = refresh_tokens(old_refresh_token, SECRET)

assert second_refresh.success is False

assert second_refresh.error == "already_used"

# --- 불변 조건 테스트 2: 탈취 시도 시 모든 세션 종료 ---

def test_invariant_session_revoked_on_reuse(fake_redis):

"""

이미 사용된 토큰으로 재시도하면 해당 사용자의 모든 세션이 종료됩니다.

새로 발급받은 토큰으로도 갱신이 불가능해야 합니다.

"""

issue_result = issue_tokens("user_003", SECRET)

old_token = issue_result.tokens.refresh_token

# 정상 갱신으로 새 토큰 획득

first_refresh = refresh_tokens(old_token, SECRET)

new_token = first_refresh.tokens.refresh_token

# 공격자가 이전 토큰으로 재시도 → 세션 전체 종료

refresh_tokens(old_token, SECRET)

# 정상 사용자도 새 토큰으로 갱신 불가

third_refresh = refresh_tokens(new_token, SECRET)

assert third_refresh.success is False

assert third_refresh.error == "session_revoked"

# --- 불변 조건 테스트 3: 동시 요청 중 하나만 성공 ---

def test_invariant_concurrent_refresh_blocked(fake_redis):

"""

같은 리프레시 토큰으로 동시에 두 요청이 오면

하나만 성공하고 나머지는 반드시 실패해야 합니다.

"""

import threading

issue_result = issue_tokens("user_004", SECRET)

token = issue_result.tokens.refresh_token

results = []

def attempt_refresh():

result = refresh_tokens(token, SECRET)

results.append(result)

threads = [threading.Thread(target=attempt_refresh) for _ in range(5)]

for t in threads:

t.start()

for t in threads:

t.join()

success_count = sum(1 for r in results if r.success)

assert success_count == 1, f"오직 하나만 성공해야 하는데 {success_count}개가 성공했습니다"

테스트 작성으로 위험도 점수가 내려갑니다.

정확성 검증 테스트 (Round-trip) 작성: -15점
불변 조건 테스트 3개 작성: -10점

최종 위험도: 0점 ✅

배포 전 마지막으로 확인할 사항이 하나 있습니다. Redis 연결이 끊겼을 때도 테스트해보셨나요? storage_error 경로가 실제로 동작하는지 확인하는 테스트를 추가하는 것을 권장합니다.

세 번째 대화 — 추가 엣지 케이스

👤 사용자

Redis 연결 실패 테스트도 추가해주세요.

🤖 Simula

좋습니다. Redis 연결이 끊겼을 때 서버가 멈추지 않고 명확한 오류를 반환하는지 확인하는 테스트입니다.

💻 Code Implementation

# --- 경계 조건 테스트: Redis 연결 실패 시 안전하게 처리 ---
def test_storage_error_on_redis_failure():
    """
    Redis 연결이 끊겼을 때 서버가 멈추지 않고
    'storage_error'를 반환해야 합니다.
    """
    with patch("your_module.r") as mock_redis:
        mock_redis.setex.side_effect = redis.RedisError("connection refused")

        result = issue_tokens("user_005", SECRET)

        assert result.success is False
        assert result.error == "storage_error"


def test_refresh_storage_error_releases_lock():
    """
    갱신 도중 Redis 오류가 발생해도 잠금이 반드시 해제되어야 합니다.
    잠금이 남아있으면 해당 사용자는 영구적으로 갱신 불가 상태가 됩니다.
    """
    with patch("your_module.r") as mock_redis:
        mock_redis.set.return_value = True   # 잠금 획득은 성공
        mock_redis.get.side_effect = redis.RedisError("connection lost")

        result = refresh_tokens("any.valid.token", SECRET)

        assert result.success is False
        # 잠금 해제가 호출됐는지 확인
        mock_redis.delete.assert_called()