xiaoxia-saas/tests/unit/test_jwt_handler.py

"""JWT Handler 单元测试."""

from __future__ import annotations

import time

import pytest
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError

from packages.application.auth.jwt_handler import (
    JWTHandler,
    configure_jwt_handler,
    get_jwt_handler,
)


@pytest.fixture
def jwt_handler():
    return JWTHandler(
        secret_key="test-secret-key-12345",
        algorithm="HS256",
        access_token_expire_minutes=30,
    )


class TestJWTHandler:
    """JWTHandler 测试"""

    def test_create_access_token_returns_string(self, jwt_handler):
        """创建 access_token 返回非空字符串"""
        token = jwt_handler.create_access_token(user_id="user_001")

        assert isinstance(token, str)
        assert len(token) > 0

    def test_create_access_token_with_role(self, jwt_handler):
        """创建带 role 的 access_token"""
        token = jwt_handler.create_access_token(user_id="user_001", role="admin")
        payload = jwt_handler.verify_access_token(token)

        assert payload["sub"] == "user_001"
        assert payload["role"] == "admin"

    def test_create_access_token_with_additional_claims(self, jwt_handler):
        """创建带额外声明的 access_token"""
        token = jwt_handler.create_access_token(
            user_id="user_001",
            additional_claims={"email": "test@example.com", "tenant": "t1"},
        )
        payload = jwt_handler.verify_access_token(token)

        assert payload["sub"] == "user_001"
        assert payload["email"] == "test@example.com"
        assert payload["tenant"] == "t1"

    def test_verify_access_token_success(self, jwt_handler):
        """验证有效 access_token"""
        token = jwt_handler.create_access_token(user_id="user_001")
        payload = jwt_handler.verify_access_token(token)

        assert payload["sub"] == "user_001"
        assert "exp" in payload
        assert "iat" in payload

    def test_verify_access_token_type_check(self, jwt_handler):
        """verify_access_token 验证 token 类型为 access"""
        token = jwt_handler.create_access_token(user_id="user_001")
        payload = jwt_handler.verify_access_token(token)

        assert payload.get("type") == "access" or "type" in payload

    def test_verify_token_no_type_restriction(self, jwt_handler):
        """verify_token 不限制 token 类型"""
        token = jwt_handler.create_access_token(user_id="user_001")
        payload = jwt_handler.verify_token(token)

        assert payload["sub"] == "user_001"

    def test_expired_token_raises_error(self):
        """过期 token 验证失败"""
        handler = JWTHandler(
            secret_key="test-secret",
            access_token_expire_minutes=-1,  # 立即过期
        )
        token = handler.create_access_token(user_id="user_001")

        # 等待一小段时间确保过期
        time.sleep(0.1)

        with pytest.raises(ExpiredSignatureError):
            handler.verify_access_token(token)

    def test_invalid_token_raises_error(self, jwt_handler):
        """无效 token 验证失败"""
        with pytest.raises(InvalidTokenError):
            jwt_handler.verify_access_token("invalid.token.here")

    def test_empty_token_raises_error(self, jwt_handler):
        """空字符串 token 验证失败"""
        with pytest.raises(InvalidTokenError):
            jwt_handler.verify_access_token("")

    def test_different_secret_fails_verification(self):
        """不同密钥生成的 token 无法互相验证"""
        handler1 = JWTHandler(secret_key="secret-one")
        handler2 = JWTHandler(secret_key="secret-two")

        token = handler1.create_access_token(user_id="user_001")

        with pytest.raises(InvalidTokenError):
            handler2.verify_access_token(token)

    def test_custom_algorithm(self):
        """支持自定义算法"""
        handler = JWTHandler(
            secret_key="test-secret",
            algorithm="HS256",
        )
        token = handler.create_access_token(user_id="user_001")
        payload = handler.verify_access_token(token)

        assert payload["sub"] == "user_001"

    def test_default_role_is_empty_string(self, jwt_handler):
        """不传 role 时默认为空字符串"""
        token = jwt_handler.create_access_token(user_id="user_001")
        payload = jwt_handler.verify_access_token(token)

        assert payload.get("role", "") == ""


class TestGlobalJWTHandler:
    """全局 JWT handler 配置测试"""

    def test_configure_creates_handler(self):
        """configure_jwt_handler 创建并返回 handler"""
        import packages.application.auth.jwt_handler as jwt_module

        # 重置全局状态
        jwt_module._default_handler = None

        handler = configure_jwt_handler(
            secret_key="global-secret",
            access_token_expire_minutes=60,
        )

        assert isinstance(handler, JWTHandler)
        assert get_jwt_handler() is handler

    def test_get_jwt_handler_without_config_raises(self):
        """未配置时调用 get_jwt_handler 抛出 RuntimeError"""
        import packages.application.auth.jwt_handler as jwt_module

        # 重置全局状态
        jwt_module._default_handler = None

        with pytest.raises(RuntimeError, match="JWT handler not configured"):
            get_jwt_handler()

    def test_configure_overwrites_existing(self):
        """重新配置会覆盖之前的 handler"""
        import packages.application.auth.jwt_handler as jwt_module

        jwt_module._default_handler = None

        handler1 = configure_jwt_handler(secret_key="first-secret")
        handler2 = configure_jwt_handler(secret_key="second-secret")

        assert handler1 is not handler2
        assert get_jwt_handler() is handler2