第4章 过期自动续期测试

4.1 双模型原理

自动续期的核心是双模型：一个短期有效的访问令牌（Access ）用于资源请求，一个长期有效的刷新令牌（Refresh ）用于获取新的访问令牌。当访问令牌过期时，客户端使用刷新令牌向认证服务请求新令牌，再重试原请求。

4.2 自动续期流程步骤

1. 初始认证：客户端提交凭证，服务端返回Access 和Refresh 。
2. 正常请求：携Access 请求资源，服务端返回200。
3. 过期检测：客户端收到401或预先解析JWT的 exp 字段判断剩余时间。
4. 发起刷新：调用刷新端点，携带Refresh （及可能的CSRF ）。
5. 服务端验证：校验Refresh 有效性、是否被撤销，颁发新Access （可选旋转Refresh ）。
6. 重试原请求：使用新Access 重试失败队列中的请求。
7. 处理刷新失败：若Refresh 也过期或无效，则强制用户重新登录。

4.3 代码实现：自动刷新拦截器

以下Python示例使用 requests.Session 实现自动刷新和请求重试，附注释说明过期检测点。

import time
import requests
from threading import Lock

class AutoRefreshSession:
    def __init__(self, auth_url: str, refresh_url: str, credentials: dict):
        self.session = requests.Session()
        self.auth_url = auth_url
        self.refresh_url = refresh_url
        self.credentials = credentials
        self.access_ = None
        self.refresh_ = None
        self.expires_at = 0   # Unix timestamp
        self._lock = Lock()

    def authenticate(self):
        """首次登录获取双"""
        resp = self.session.post(self.auth_url, json=self.credentials)
        resp.raise_for_status()
        data = resp.json()
        self.access_ = data['access_']
        self.refresh_ = data['refresh_']
        # 假设JWT解析或服务直接返回exp
        self.expires_at = time.time() + data.get('expires_in', 900) - 5

    def _is_expired(self) -> bool:
        return time.time() >= self.expires_at

    def _refresh(self):
        """使用Refresh 获取新Access """
        with self._lock:  # 防止并发刷新
            if not self._is_expired():  # 双重检查
                return
            resp = self.session.post(self.refresh_url, json={
                'refresh_': self.refresh_
            })
            if resp.status_code == 401:
                # Refresh 失效，重新登录
                self.authenticate()
                return
            resp.raise_for_status()
            data = resp.json()
            self.access_ = data['access_']
            # 如果服务端旋转Refresh ，也更新
            if 'refresh_' in data:
                self.refresh_ = data['refresh_']
            self.expires_at = time.time() + data.get('expires_in', 900) - 5

    def request(self, method, url, **kwargs):
        # 每次请求前检查并刷新
        if self._is_expired():
            self._refresh()
        headers = kwargs.get('headers', {})
        headers['Authorization'] = f'Bearer {self.access_}'
        kwargs['headers'] = headers
        resp = self.session.request(method, url, **kwargs)
        # 服务端返回401时也可能触发重试
        if resp.status_code == 401 and not getattr(resp, '_retried', False):
            self._refresh()
            headers['Authorization'] = f'Bearer {self.access_}'
            kwargs['headers'] = headers
            resp = self.session.request(method, url, **kwargs)
            resp._retried = True
        return resp

4.4 测试用例设计

使用 unittest 编写测试，利用mock模拟服务端响应时间。

import unittest
from unittest.mock import patch, Mock
from freezegun import freeze_time

class TestAutoRefresh(unittest.TestCase):
    def setUp(self):
        self.session = AutoRefreshSession('http://auth/login', 'http://auth/refresh', {'user':'test','pass':'x'})
        # mock首次登录
        with patch.object(self.session.session, 'post') as mock_post:
            mock_post.return_value.status_code = 200
            mock_post.return_value.json.return_value = {
                'access_': 'old_at',
                'refresh_': 'old_rt',
                'expires_in': 900
            }
            self.session.authenticate()

    @freeze_time("2025-01-01 00:00:00")
    def test_auto_refresh_before_expiry(self):
        """提前5秒过期检测，自动触发刷新"""
        with patch.object(self.session.session, 'post') as mock_post:
            # 刷新端点返回新
            mock_post.return_value.status_code = 200
            mock_post.return_value.json.return_value = {
                'access_': 'new_at',
                'refresh_': 'old_rt',  # 不旋转
                'expires_in': 900
            }
            # 模拟时间走到过期前4秒（触发检测）
            with freeze_time("2025-01-01 00:14:56"):
                resp = self.session.request('GET', 'http://resource/api')
                self.assertEqual(resp.request.headers['Authorization'], 'Bearer new_at')

4.5 常见错误与优化

4.6 性能与稳定性优化建议

• 缓存Access ：避免每次请求都解析JWT或查询数据库，使用本地内存缓存。
• 预测性刷新：在过期前几秒提前异步刷新，减少阻塞。
• 降级策略：当刷新端点连续失败（如3次），等待指数退避再重试，避免雪崩。
• 日志监控：记录刷新次数、失败率，告警异常（如频繁刷新、旋转失效）。
• 无状态设计：服务端Refresh 使用自包含JWT或数据库标记撤销，避免session存储瓶颈。