Xiaoxia AI
133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
"""
|
|
JWT 工具类测试
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
|
|
|
|
from packages.domain.auth.jwt_service import JWTConfig, JWTService, TokenType
|
|
|
|
|
|
class TestJWTService:
|
|
"""JWT 服务测试"""
|
|
|
|
@pytest.fixture
|
|
def jwt_service(self):
|
|
"""创建 JWT 服务实例"""
|
|
config = JWTConfig()
|
|
config.SECRET_KEY = "test-secret-key-for-testing"
|
|
return JWTService(config)
|
|
|
|
def test_create_access_token(self, jwt_service):
|
|
"""测试创建 access_token"""
|
|
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
|
|
|
|
assert isinstance(token, str)
|
|
assert len(token) > 0
|
|
|
|
# 验证 Token 内容
|
|
payload = jwt_service.verify_access_token(token)
|
|
assert payload["sub"] == "user-123"
|
|
assert payload["workspace_id"] == "workspace-456"
|
|
assert payload["role"] == "admin"
|
|
assert payload["type"] == TokenType.ACCESS
|
|
|
|
def test_create_refresh_token(self, jwt_service):
|
|
"""测试创建 refresh_token"""
|
|
token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
|
|
|
|
assert isinstance(token, str)
|
|
assert len(token) > 0
|
|
|
|
# 验证 Token 内容
|
|
payload = jwt_service.verify_refresh_token(token)
|
|
assert payload["sub"] == "user-123"
|
|
assert payload["session_id"] == "session-789"
|
|
assert payload["type"] == TokenType.REFRESH
|
|
|
|
def test_verify_valid_access_token(self, jwt_service):
|
|
"""测试验证有效的 access_token"""
|
|
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="member")
|
|
|
|
payload = jwt_service.verify_access_token(token)
|
|
assert payload["sub"] == "user-123"
|
|
assert payload["workspace_id"] == "workspace-456"
|
|
assert payload["role"] == "member"
|
|
|
|
def test_verify_expired_token(self, jwt_service):
|
|
"""测试验证过期的 Token"""
|
|
# 创建一个已过期的配置（使用相同的 SECRET_KEY）
|
|
config = JWTConfig()
|
|
config.SECRET_KEY = "test-secret-key-for-testing" # 与 fixture 相同
|
|
config.ACCESS_TOKEN_EXPIRE_MINUTES = -1 # 负数，立即过期
|
|
|
|
expired_service = JWTService(config)
|
|
token = expired_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
|
|
|
|
# 验证应该抛出过期异常
|
|
with pytest.raises(ExpiredSignatureError):
|
|
jwt_service.verify_access_token(token)
|
|
|
|
def test_verify_invalid_token(self, jwt_service):
|
|
"""测试验证无效的 Token"""
|
|
invalid_token = "invalid.token.string"
|
|
|
|
with pytest.raises(InvalidTokenError):
|
|
jwt_service.verify_access_token(invalid_token)
|
|
|
|
def test_verify_wrong_token_type(self, jwt_service):
|
|
"""测试验证错误类型的 Token"""
|
|
# 创建 refresh_token
|
|
refresh_token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
|
|
|
|
# 用 verify_access_token 验证应该失败
|
|
with pytest.raises(ValueError, match="Token type must be 'access'"):
|
|
jwt_service.verify_access_token(refresh_token)
|
|
|
|
# 反过来也一样
|
|
access_token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
|
|
|
|
with pytest.raises(ValueError, match="Token type must be 'refresh'"):
|
|
jwt_service.verify_refresh_token(access_token)
|
|
|
|
def test_verify_tampered_token(self, jwt_service):
|
|
"""测试验证被篡改的 Token"""
|
|
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
|
|
|
|
# 篡改 Token（修改最后几个字符）
|
|
tampered_token = token[:-5] + "XXXXX"
|
|
|
|
with pytest.raises(InvalidTokenError):
|
|
jwt_service.verify_access_token(tampered_token)
|
|
|
|
def test_additional_claims(self, jwt_service):
|
|
"""测试额外的声明"""
|
|
token = jwt_service.create_access_token(
|
|
user_id="user-123",
|
|
workspace_id="workspace-456",
|
|
role="admin",
|
|
additional_claims={
|
|
"email": "user@example.com",
|
|
"display_name": "Test User",
|
|
},
|
|
)
|
|
|
|
payload = jwt_service.verify_access_token(token)
|
|
assert payload["email"] == "user@example.com"
|
|
assert payload["display_name"] == "Test User"
|
|
|
|
def test_decode_unsafe(self, jwt_service):
|
|
"""测试不安全解码（不验证签名）"""
|
|
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
|
|
|
|
# 不验证签名地解码
|
|
payload = jwt_service.decode_token_unsafe(token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "user-123"
|
|
|
|
# 无效 Token 应该返回 None
|
|
invalid_payload = jwt_service.decode_token_unsafe("invalid.token")
|
|
assert invalid_payload is None
|