xiaoxia-saas/tests/unit/test_jwt_service.py

"""
JWT Service 单元测试
"""

import time
from datetime import datetime, timedelta

import jwt
import pytest
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError

from packages.application.auth.jwt_service import (
    JWTConfig,
    JWTService,
    TokenType,
)


class TestJWTConfig:
    """JWT 配置测试"""

    def test_config_init_success(self):
        """测试正常初始化"""
        config = JWTConfig(secret_key="a-very-strong-secret-key-for-testing")
        assert config.SECRET_KEY == "a-very-strong-secret-key-for-testing"
        assert config.ALGORITHM == "HS256"
        assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 15
        assert config.REFRESH_TOKEN_EXPIRE_DAYS == 7

    def test_config_custom_values(self):
        """测试自定义配置值"""
        config = JWTConfig(
            secret_key="test-secret",
            algorithm="HS512",
            access_token_expire_minutes=30,
            refresh_token_expire_days=14,
        )
        assert config.ALGORITHM == "HS512"
        assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 30
        assert config.REFRESH_TOKEN_EXPIRE_DAYS == 14

    def test_config_empty_secret_raises(self):
        """测试空密钥报错"""
        with pytest.raises(ValueError, match="secret_key must be provided"):
            JWTConfig(secret_key="")

    def test_config_whitespace_secret_raises(self):
        """测试全空格密钥报错"""
        with pytest.raises(ValueError, match="secret_key must be provided"):
            JWTConfig(secret_key="   ")

    def test_config_insecure_default_secret_raises(self):
        """测试不安全的默认密钥报错"""
        insecure_keys = [
            "your-secret-key-change-in-production",
            "your-secret-key",
            "secret",
            "changeme",
            "password",
            "YOUR-SECRET-KEY",
            "Secret",
        ]
        for key in insecure_keys:
            with pytest.raises(ValueError, match="insecure"):
                JWTConfig(secret_key=key)


class TestJWTService:
    """JWT 服务测试"""

    @pytest.fixture
    def config(self):
        return JWTConfig(
            secret_key="test-secret-key-for-jwt-unit-tests-12345",
            algorithm="HS256",
            access_token_expire_minutes=30,
            refresh_token_expire_days=7,
        )

    @pytest.fixture
    def service(self, config):
        return JWTService(config=config)

    def test_service_init_without_config_raises(self):
        """测试无 config 初始化报错"""
        with pytest.raises(ValueError, match="requires a JWTConfig"):
            JWTService(config=None)

    # --- create_access_token ---

    def test_create_access_token_success(self, service):
        """测试创建 access token 成功"""
        token = service.create_access_token(user_id="user-123")
        assert isinstance(token, str)
        assert len(token) > 0

    def test_create_access_token_contains_user_id(self, service, config):
        """测试 access token 包含正确的 user_id"""
        token = service.create_access_token(user_id="user-456")
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert payload["sub"] == "user-456"

    def test_create_access_token_has_correct_type(self, service, config):
        """测试 access token 类型正确"""
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert payload["type"] == TokenType.ACCESS

    def test_create_access_token_contains_role(self, service, config):
        """测试 access token 包含角色"""
        token = service.create_access_token(user_id="user-123", role="admin")
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert payload["role"] == "admin"

    def test_create_access_token_additional_claims(self, service, config):
        """测试 access token 包含额外声明"""
        token = service.create_access_token(
            user_id="user-123",
            additional_claims={"custom_field": "custom_value", "sid": "session-abc"},
        )
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert payload["custom_field"] == "custom_value"
        assert payload["sid"] == "session-abc"

    def test_create_access_token_has_iat_and_exp(self, service, config):
        """测试 access token 包含 iat 和 exp"""
        before = datetime.utcnow() - timedelta(seconds=1)
        token = service.create_access_token(user_id="user-123")
        after = datetime.utcnow() + timedelta(seconds=1)

        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert "iat" in payload
        assert "exp" in payload

        iat = datetime.utcfromtimestamp(payload["iat"])
        exp = datetime.utcfromtimestamp(payload["exp"])

        assert before <= iat <= after
        assert exp > iat
        # 过期时间约等于配置的分钟数
        expected_expiry = timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES)
        actual_expiry = exp - iat
        assert abs((actual_expiry - expected_expiry).total_seconds()) < 5

    # --- create_refresh_token ---

    def test_create_refresh_token_success(self, service):
        """测试创建 refresh token 成功"""
        token = service.create_refresh_token(user_id="user-123", session_id="sess-abc")
        assert isinstance(token, str)
        assert len(token) > 0

    def test_create_refresh_token_contains_correct_data(self, service, config):
        """测试 refresh token 包含正确数据"""
        token = service.create_refresh_token(user_id="user-789", session_id="sess-xyz")
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        assert payload["sub"] == "user-789"
        assert payload["session_id"] == "sess-xyz"
        assert payload["type"] == TokenType.REFRESH

    def test_create_refresh_token_expiry(self, service, config):
        """测试 refresh token 过期时间正确"""
        token = service.create_refresh_token(user_id="user-123", session_id="sess-abc")
        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])

        iat = datetime.utcfromtimestamp(payload["iat"])
        exp = datetime.utcfromtimestamp(payload["exp"])
        expected_expiry = timedelta(days=config.REFRESH_TOKEN_EXPIRE_DAYS)
        actual_expiry = exp - iat
        assert abs((actual_expiry - expected_expiry).total_seconds()) < 5

    # --- verify_token ---

    def test_verify_valid_token(self, service):
        """测试验证有效 token"""
        token = service.create_access_token(user_id="user-123")
        payload = service.verify_token(token)
        assert payload["sub"] == "user-123"

    def test_verify_expired_token_raises(self, service, config):
        """测试验证过期 token 报错"""
        # 创建一个已经过期的 token
        payload = {
            "sub": "user-123",
            "type": TokenType.ACCESS,
            "iat": datetime.utcnow() - timedelta(hours=1),
            "exp": datetime.utcnow() - timedelta(minutes=30),
        }
        expired_token = jwt.encode(payload, config.SECRET_KEY, algorithm=config.ALGORITHM)

        with pytest.raises(ExpiredSignatureError, match="expired"):
            service.verify_token(expired_token)

    def test_verify_invalid_token_raises(self, service):
        """测试验证无效 token 报错"""
        with pytest.raises(InvalidTokenError):
            service.verify_token("this-is-not-a-valid-jwt-token")

    def test_verify_token_with_wrong_secret_raises(self, service, config):
        """测试用错误密钥签发的 token 验证失败"""
        wrong_config = JWTConfig(secret_key="different-secret-key")
        wrong_service = JWTService(config=wrong_config)
        token = wrong_service.create_access_token(user_id="user-123")

        with pytest.raises(InvalidTokenError):
            service.verify_token(token)

    # --- verify_access_token ---

    def test_verify_access_token_success(self, service):
        """测试验证有效的 access token"""
        token = service.create_access_token(user_id="user-123", role="user")
        payload = service.verify_access_token(token)
        assert payload["sub"] == "user-123"
        assert payload["type"] == TokenType.ACCESS

    def test_verify_access_token_with_refresh_token_raises(self, service):
        """测试用 refresh token 调用 verify_access_token 报错"""
        refresh_token = service.create_refresh_token(user_id="user-123", session_id="sess-abc")

        with pytest.raises(ValueError, match="Token type must be 'access'"):
            service.verify_access_token(refresh_token)

    # --- verify_refresh_token ---

    def test_verify_refresh_token_success(self, service):
        """测试验证有效的 refresh token"""
        token = service.create_refresh_token(user_id="user-123", session_id="sess-abc")
        payload = service.verify_refresh_token(token)
        assert payload["sub"] == "user-123"
        assert payload["session_id"] == "sess-abc"
        assert payload["type"] == TokenType.REFRESH

    def test_verify_refresh_token_with_access_token_raises(self, service):
        """测试用 access token 调用 verify_refresh_token 报错"""
        access_token = service.create_access_token(user_id="user-123")

        with pytest.raises(ValueError, match="Token type must be 'refresh'"):
            service.verify_refresh_token(access_token)

    def test_access_and_refresh_tokens_are_different(self, service):
        """测试 access token 和 refresh token 不相同"""
        access = service.create_access_token(user_id="user-123")
        refresh = service.create_refresh_token(user_id="user-123", session_id="sess-abc")
        assert access != refresh