xiaoxia-saas/tests/unit/test_jwt_service.py

"""JWT 服务与处理器单元测试."""

import time
from datetime import datetime, timedelta, timezone

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

from packages.application.auth.jwt_handler import (
    JWTHandler,
    configure_jwt_handler,
    get_jwt_handler,
)
from packages.application.auth.jwt_service import (
    JWTConfig,
    JWTService,
    TokenType,
)

# ── 测试常量 ──────────────────────────────────────────────────────────────────

TEST_SECRET = "test-secret-key-for-unit-testing-only-not-for-production"
STRONG_SECRET = "x" * 32  # 满足长度要求的测试密钥


# ── JWTConfig 测试 ───────────────────────────────────────────────────────────


class TestJWTConfig:
    """JWTConfig 配置类测试"""

    def test_init_with_valid_secret(self):
        config = JWTConfig(secret_key=STRONG_SECRET)
        assert config.SECRET_KEY == STRONG_SECRET
        assert config.ALGORITHM == "HS256"
        assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 15
        assert config.REFRESH_TOKEN_EXPIRE_DAYS == 7

    def test_init_custom_values(self):
        config = JWTConfig(
            secret_key=STRONG_SECRET,
            algorithm="HS384",
            access_token_expire_minutes=60,
            refresh_token_expire_days=14,
        )
        assert config.SECRET_KEY == STRONG_SECRET
        assert config.ALGORITHM == "HS384"
        assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 60
        assert config.REFRESH_TOKEN_EXPIRE_DAYS == 14

    def test_empty_secret_raises(self):
        with pytest.raises(ValueError, match="secret_key must be provided"):
            JWTConfig(secret_key="")

    def test_whitespace_only_secret_raises(self):
        with pytest.raises(ValueError, match="secret_key must be provided"):
            JWTConfig(secret_key="   ")

    def test_none_secret_raises(self):
        with pytest.raises(ValueError, match="secret_key must be provided"):
            JWTConfig(secret_key=None)

    @pytest.mark.parametrize(
        "insecure_secret",
        [
            "your-secret-key-change-in-production",
            "your-secret-key",
            "secret",
            "changeme",
            "password",
            "SECRET",
            "Your-Secret-Key",
        ],
    )
    def test_insecure_default_secret_raises(self, insecure_secret):
        with pytest.raises(ValueError, match="insecure"):
            JWTConfig(secret_key=insecure_secret)

    def test_zero_expire_minutes_allowed(self):
        config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
        assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 0

    def test_negative_expire_days_allowed(self):
        # 配置类不校验合理性，由业务层判断
        config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=-1)
        assert config.REFRESH_TOKEN_EXPIRE_DAYS == -1


# ── JWTService 初始化测试 ────────────────────────────────────────────────────


class TestJWTServiceInit:
    """JWTService 初始化测试"""

    def test_init_with_config(self):
        config = JWTConfig(secret_key=STRONG_SECRET)
        service = JWTService(config)
        assert service.config is config

    def test_init_none_config_raises(self):
        with pytest.raises(ValueError, match="JWTService requires a JWTConfig"):
            JWTService(None)


# ── TokenType 测试 ───────────────────────────────────────────────────────────


class TestTokenType:
    """TokenType 常量测试"""

    def test_access_value(self):
        assert TokenType.ACCESS == "access"

    def test_refresh_value(self):
        assert TokenType.REFRESH == "refresh"

    def test_access_and_refresh_different(self):
        assert TokenType.ACCESS != TokenType.REFRESH


# ── JWTService create_access_token 测试 ─────────────────────────────────────


class TestCreateAccessToken:
    """创建 access_token 测试"""

    @pytest.fixture
    def service(self):
        return JWTService(JWTConfig(secret_key=STRONG_SECRET))

    def test_creates_valid_jwt_string(self, service):
        token = service.create_access_token(user_id="user-123")
        assert isinstance(token, str)
        assert len(token) > 0

    def test_token_contains_user_id_as_sub(self, service):
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["sub"] == "user-123"

    def test_token_type_is_access(self, service):
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["type"] == TokenType.ACCESS

    def test_default_role_is_empty_string(self, service):
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["role"] == ""

    def test_custom_role(self, service):
        token = service.create_access_token(user_id="user-123", role="admin")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["role"] == "admin"

    def test_has_iat_and_exp(self, service):
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert "iat" in payload
        assert "exp" in payload
        assert payload["exp"] > payload["iat"]

    def test_expire_matches_config(self, service):
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        iat = datetime.fromtimestamp(payload["iat"], tz=timezone.utc)
        exp = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
        delta = exp - iat
        assert delta.total_seconds() == 15 * 60  # 15分钟

    def test_custom_expire_time(self):
        config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=30)
        service = JWTService(config)
        token = service.create_access_token(user_id="user-123")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        delta = payload["exp"] - payload["iat"]
        assert delta == 30 * 60

    def test_additional_claims(self, service):
        extra = {"custom_field": "value", "another": 42}
        token = service.create_access_token(user_id="user-123", additional_claims=extra)
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["custom_field"] == "value"
        assert payload["another"] == 42

    def test_additional_claims_can_override_standard(self, service):
        # additional_claims 可以覆盖标准字段（由调用者负责）
        token = service.create_access_token(
            user_id="user-123",
            additional_claims={"sub": "overridden"},
        )
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["sub"] == "overridden"

    def test_additional_claims_none_is_same_as_empty(self, service):
        token = service.create_access_token(user_id="user-123", additional_claims=None)
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["sub"] == "user-123"

    def test_uses_correct_algorithm(self):
        config = JWTConfig(secret_key=STRONG_SECRET, algorithm="HS384")
        service = JWTService(config)
        token = service.create_access_token(user_id="u1")
        # 用 HS256 解码应该失败
        with pytest.raises(InvalidTokenError):
            jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        # 用 HS384 解码应该成功
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS384"])
        assert payload["sub"] == "u1"


# ── JWTService create_refresh_token 测试 ────────────────────────────────────


class TestCreateRefreshToken:
    """创建 refresh_token 测试"""

    @pytest.fixture
    def service(self):
        return JWTService(JWTConfig(secret_key=STRONG_SECRET))

    def test_creates_valid_string(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        assert isinstance(token, str)
        assert len(token) > 0

    def test_contains_user_id_and_session_id(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="sess-abc")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["sub"] == "u1"
        assert payload["session_id"] == "sess-abc"

    def test_token_type_is_refresh(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert payload["type"] == TokenType.REFRESH

    def test_has_iat_and_exp(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        assert "iat" in payload
        assert "exp" in payload
        assert payload["exp"] > payload["iat"]

    def test_expire_matches_config_days(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        delta = payload["exp"] - payload["iat"]
        assert delta == 7 * 24 * 60 * 60  # 7天

    def test_custom_refresh_expire_days(self):
        config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=30)
        service = JWTService(config)
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
        delta = payload["exp"] - payload["iat"]
        assert delta == 30 * 24 * 60 * 60


# ── JWTService verify_token 测试 ────────────────────────────────────────────


class TestVerifyToken:
    """通用 Token 验证测试"""

    @pytest.fixture
    def service(self):
        return JWTService(JWTConfig(secret_key=STRONG_SECRET))

    def test_verify_valid_access_token(self, service):
        token = service.create_access_token(user_id="u1")
        payload = service.verify_token(token)
        assert payload["sub"] == "u1"
        assert payload["type"] == TokenType.ACCESS

    def test_verify_valid_refresh_token(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = service.verify_token(token)
        assert payload["sub"] == "u1"
        assert payload["session_id"] == "s1"

    def test_verify_expired_token_raises(self, service):
        config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
        svc = JWTService(config)
        token = svc.create_access_token(user_id="u1")
        # 0 分钟过期，立即过期
        time.sleep(0.1)  # 稍微等一下确保过期
        with pytest.raises(ExpiredSignatureError):
            svc.verify_token(token)

    def test_verify_wrong_secret_raises(self, service):
        token = service.create_access_token(user_id="u1")
        other_service = JWTService(JWTConfig(secret_key="different-secret-1234567890"))
        with pytest.raises(InvalidTokenError):
            other_service.verify_token(token)

    def test_verify_tampered_token_raises(self, service):
        token = service.create_access_token(user_id="u1")
        # 篡改 token 中间部分
        parts = token.split(".")
        assert len(parts) == 3
        tampered = parts[0] + "." + parts[1][:-1] + "A." + parts[2]
        with pytest.raises(InvalidTokenError):
            service.verify_token(tampered)

    def test_verify_empty_string_raises(self, service):
        with pytest.raises(InvalidTokenError):
            service.verify_token("")

    def test_verify_garbage_string_raises(self, service):
        with pytest.raises(InvalidTokenError):
            service.verify_token("not.a.valid.jwt.token")

    def test_verify_returns_dict(self, service):
        token = service.create_access_token(user_id="u1", role="admin")
        payload = service.verify_token(token)
        assert isinstance(payload, dict)
        assert "sub" in payload
        assert "role" in payload


# ── JWTService verify_access_token 测试 ─────────────────────────────────────


class TestVerifyAccessToken:
    """Access Token 专属验证测试"""

    @pytest.fixture
    def service(self):
        return JWTService(JWTConfig(secret_key=STRONG_SECRET))

    def test_valid_access_token_passes(self, service):
        token = service.create_access_token(user_id="u1", role="admin")
        payload = service.verify_access_token(token)
        assert payload["sub"] == "u1"
        assert payload["role"] == "admin"

    def test_refresh_token_fails_type_check(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        with pytest.raises(ValueError, match="Token type must be 'access'"):
            service.verify_access_token(token)

    def test_token_without_type_field_raises(self, service):
        # 手动构造一个没有 type 字段的 token
        payload_data = {"sub": "u1", "iat": 1000, "exp": 9999999999}
        token = jwt.encode(payload_data, STRONG_SECRET, algorithm="HS256")
        with pytest.raises(ValueError, match="Token type must be 'access'"):
            service.verify_access_token(token)

    def test_expired_access_token_raises_expired_error(self, service):
        config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
        svc = JWTService(config)
        token = svc.create_access_token(user_id="u1")
        time.sleep(0.1)
        with pytest.raises(ExpiredSignatureError):
            svc.verify_access_token(token)


# ── JWTService verify_refresh_token 测试 ────────────────────────────────────


class TestVerifyRefreshToken:
    """Refresh Token 专属验证测试"""

    @pytest.fixture
    def service(self):
        return JWTService(JWTConfig(secret_key=STRONG_SECRET))

    def test_valid_refresh_token_passes(self, service):
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        payload = service.verify_refresh_token(token)
        assert payload["sub"] == "u1"
        assert payload["session_id"] == "s1"

    def test_access_token_fails_type_check(self, service):
        token = service.create_access_token(user_id="u1")
        with pytest.raises(ValueError, match="Token type must be 'refresh'"):
            service.verify_refresh_token(token)

    def test_token_without_type_field_raises(self, service):
        payload_data = {"sub": "u1", "session_id": "s1", "iat": 1000, "exp": 9999999999}
        token = jwt.encode(payload_data, STRONG_SECRET, algorithm="HS256")
        with pytest.raises(ValueError, match="Token type must be 'refresh'"):
            service.verify_refresh_token(token)

    def test_expired_refresh_token_raises(self):
        config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=0)
        service = JWTService(config)
        token = service.create_refresh_token(user_id="u1", session_id="s1")
        # 0天过期，应该立即使exp <= iat
        with pytest.raises(ExpiredSignatureError):
            service.verify_refresh_token(token)


# ── JWTHandler 委托层测试 ────────────────────────────────────────────────────


class TestJWTHandler:
    """JWTHandler 委托层测试"""

    def test_init_creates_handler(self):
        handler = JWTHandler(secret_key=STRONG_SECRET)
        assert handler is not None

    def test_create_and_verify_access_token(self):
        handler = JWTHandler(secret_key=STRONG_SECRET)
        token = handler.create_access_token(user_id="u1", role="user")
        payload = handler.verify_access_token(token)
        assert payload["sub"] == "u1"
        assert payload["role"] == "user"

    def test_verify_token_generic(self):
        handler = JWTHandler(secret_key=STRONG_SECRET)
        token = handler.create_access_token(user_id="u1")
        payload = handler.verify_token(token)
        assert payload["sub"] == "u1"

    def test_custom_algorithm(self):
        handler = JWTHandler(secret_key=STRONG_SECRET, algorithm="HS384")
        token = handler.create_access_token(user_id="u1")
        payload = handler.verify_access_token(token)
        assert payload["sub"] == "u1"

    def test_custom_expire_minutes(self):
        handler = JWTHandler(secret_key=STRONG_SECRET, access_token_expire_minutes=45)
        token = handler.create_access_token(user_id="u1")
        payload = handler.verify_access_token(token)
        delta = payload["exp"] - payload["iat"]
        assert delta == 45 * 60

    def test_additional_claims_passthrough(self):
        handler = JWTHandler(secret_key=STRONG_SECRET)
        extra = {"org_id": "org-1", "plan": "pro"}
        token = handler.create_access_token("u1", additional_claims={"org_id": "org-1"})
        payload = handler.verify_access_token(
            token := handler.create_access_token("u1", additional_claims={"org_id": "org-1"})
        )
        # 这里直接测试更简洁
        payload = handler.verify_access_token(handler.create_access_token("u1", additional_claims={"x": 1}))
        assert payload["x"] == 1


# ── 全局 JWT handler 测试 ───────────────────────────────────────────────────


class TestGlobalJWTHandler:
    """全局 JWT Handler 配置与获取测试"""

    def test_configure_creates_handler(self):
        handler = configure_jwt_handler(secret_key=STRONG_SECRET)
        assert isinstance(handler, JWTHandler)

    def test_get_after_configure_works(self):
        configure_jwt_handler(secret_key=STRONG_SECRET)
        handler = get_jwt_handler()
        assert isinstance(handler, JWTHandler)
        token = handler.create_access_token(user_id="u1")
        payload = handler.verify_access_token(token)
        assert payload["sub"] == "u1"

    def test_get_before_configure_raises(self):
        # 重置全局状态（通过设置 None 模拟未配置）
        import packages.application.auth.jwt_handler as mod

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

    def test_configure_returns_same_as_get(self):
        h1 = configure_jwt_handler(secret_key=STRONG_SECRET)
        h2 = get_jwt_handler()
        assert h1 is h2

    def test_reconfigure_replaces_handler(self):
        h1 = configure_jwt_handler(secret_key=STRONG_SECRET)
        h2 = configure_jwt_handler(secret_key=STRONG_SECRET + "_new")
        assert h1 is not h2
        assert get_jwt_handler() is h2