xiaoxia-saas/tests/unit/test_jwt_service.py

"""
JWT 工具类测试
"""
import pytest
from datetime import datetime, timedelta
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError

from packages.domain.auth.jwt_service import (
    JWTService,
    JWTConfig,
    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