c14fd21eaa
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
245 lines
8.3 KiB
Python
Executable File
245 lines
8.3 KiB
Python
Executable File
"""
|
||
JWT + Password 委托层单元测试(第二十一波)
|
||
|
||
覆盖:
|
||
- JWTHandler (create/verify/configure/get)
|
||
- PasswordHandler (hash/verify/needs_rehash/validate_strength/configure/get)
|
||
"""
|
||
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
from jwt.exceptions import InvalidTokenError
|
||
|
||
from packages.application.auth.jwt_handler import (
|
||
JWTHandler,
|
||
configure_jwt_handler,
|
||
get_jwt_handler,
|
||
)
|
||
from packages.application.auth.password_handler import (
|
||
PasswordHandler,
|
||
configure_password_handler,
|
||
get_password_handler,
|
||
)
|
||
|
||
SECRET_KEY = "test-secret-key-for-unit-testing-only-not-for-production"
|
||
|
||
|
||
# ============================================================
|
||
# JWTHandler
|
||
# ============================================================
|
||
|
||
|
||
class TestJWTHandler:
|
||
"""JWTHandler JWT 委托层"""
|
||
|
||
def test_create_and_verify_access_token(self):
|
||
"""创建并验证 access_token"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY)
|
||
token = handler.create_access_token(user_id="user-123", role="admin")
|
||
|
||
assert isinstance(token, str)
|
||
assert len(token) > 0
|
||
|
||
payload = handler.verify_access_token(token)
|
||
assert payload["sub"] == "user-123"
|
||
assert payload["role"] == "admin"
|
||
assert "exp" in payload
|
||
assert "type" in payload
|
||
assert payload["type"] == "access"
|
||
|
||
def test_create_token_with_additional_claims(self):
|
||
"""携带额外 claims"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY)
|
||
token = handler.create_access_token(
|
||
user_id="user-1",
|
||
role="user",
|
||
additional_claims={"email": "a@b.com", "org_id": "org-1"},
|
||
)
|
||
payload = handler.verify_access_token(token)
|
||
assert payload["email"] == "a@b.com"
|
||
assert payload["org_id"] == "org-1"
|
||
|
||
def test_create_token_default_role(self):
|
||
"""默认 role 为空字符串"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY)
|
||
token = handler.create_access_token(user_id="user-1")
|
||
payload = handler.verify_access_token(token)
|
||
assert payload["role"] == ""
|
||
|
||
def test_verify_generic_token(self):
|
||
"""verify_token 通用验证方法"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY)
|
||
token = handler.create_access_token(user_id="user-1")
|
||
payload = handler.verify_token(token)
|
||
assert payload["sub"] == "user-1"
|
||
|
||
def test_verify_invalid_token_raises(self):
|
||
"""无效 token 验证失败"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY)
|
||
with pytest.raises(InvalidTokenError):
|
||
handler.verify_access_token("invalid-token")
|
||
|
||
def test_verify_wrong_secret(self):
|
||
"""用不同密钥签名的 token 验证失败"""
|
||
handler1 = JWTHandler(secret_key="key-a")
|
||
handler2 = JWTHandler(secret_key="key-b")
|
||
|
||
token = handler1.create_access_token(user_id="user-1")
|
||
with pytest.raises(InvalidTokenError):
|
||
handler2.verify_access_token(token)
|
||
|
||
def test_custom_algorithm(self):
|
||
"""自定义算法"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY, algorithm="HS256")
|
||
token = handler.create_access_token(user_id="user-1")
|
||
payload = handler.verify_access_token(token)
|
||
assert payload["sub"] == "user-1"
|
||
|
||
def test_custom_expire_minutes(self):
|
||
"""自定义过期时间"""
|
||
handler = JWTHandler(secret_key=SECRET_KEY, access_token_expire_minutes=60)
|
||
token = handler.create_access_token(user_id="user-1")
|
||
payload = handler.verify_access_token(token)
|
||
assert payload["sub"] == "user-1"
|
||
|
||
|
||
# ============================================================
|
||
# JWTHandler - 全局配置
|
||
# ============================================================
|
||
|
||
|
||
class TestJWTGlobalConfig:
|
||
"""JWT 全局配置与获取"""
|
||
|
||
def test_configure_and_get(self):
|
||
"""配置后可以获取"""
|
||
handler = configure_jwt_handler(secret_key=SECRET_KEY, access_token_expire_minutes=15)
|
||
assert isinstance(handler, JWTHandler)
|
||
|
||
got = get_jwt_handler()
|
||
assert got is handler
|
||
|
||
def test_reconfigure_replaces(self):
|
||
"""重新配置会替换"""
|
||
h1 = configure_jwt_handler(secret_key="key-a")
|
||
h2 = configure_jwt_handler(secret_key="key-b")
|
||
assert h1 is not h2
|
||
assert get_jwt_handler() is h2
|
||
|
||
|
||
# ============================================================
|
||
# PasswordHandler
|
||
# ============================================================
|
||
|
||
|
||
class TestPasswordHandler:
|
||
"""PasswordHandler 密码委托层"""
|
||
|
||
def test_hash_and_verify_correct(self):
|
||
"""哈希并验证正确密码"""
|
||
handler = PasswordHandler()
|
||
hashed = handler.hash_password("MySecurePass123")
|
||
|
||
assert isinstance(hashed, str)
|
||
assert hashed != "MySecurePass123"
|
||
assert handler.verify_password("MySecurePass123", hashed) is True
|
||
|
||
def test_verify_wrong_password(self):
|
||
"""验证错误密码"""
|
||
handler = PasswordHandler()
|
||
hashed = handler.hash_password("CorrectPass123")
|
||
assert handler.verify_password("WrongPass456", hashed) is False
|
||
|
||
def test_hash_is_unique_each_time(self):
|
||
"""同密码每次哈希不同(salt)"""
|
||
handler = PasswordHandler()
|
||
h1 = handler.hash_password("SamePass123")
|
||
h2 = handler.hash_password("SamePass123")
|
||
assert h1 != h2
|
||
# 但都能验证通过
|
||
assert handler.verify_password("SamePass123", h1)
|
||
assert handler.verify_password("SamePass123", h2)
|
||
|
||
def test_needs_rehash_new_hash(self):
|
||
"""新生成的哈希不需要重新计算"""
|
||
handler = PasswordHandler()
|
||
hashed = handler.hash_password("TestPass123")
|
||
assert handler.needs_rehash(hashed) is False
|
||
|
||
def test_validate_strength_strong(self):
|
||
"""强密码校验通过"""
|
||
handler = PasswordHandler()
|
||
ok, err = handler.validate_strength("StrongPass123")
|
||
assert ok is True
|
||
assert err is None or err == ""
|
||
|
||
def test_validate_strength_too_short(self):
|
||
"""密码太短"""
|
||
handler = PasswordHandler()
|
||
ok, err = handler.validate_strength("Ab1")
|
||
assert ok is False
|
||
assert err is not None
|
||
|
||
def test_validate_strength_no_uppercase(self):
|
||
"""缺少大写字母"""
|
||
handler = PasswordHandler()
|
||
ok, err = handler.validate_strength("lowercase123")
|
||
assert ok is False
|
||
assert err is not None
|
||
|
||
def test_validate_strength_no_lowercase(self):
|
||
"""缺少小写字母"""
|
||
handler = PasswordHandler()
|
||
ok, err = handler.validate_strength("UPPERCASE123")
|
||
assert ok is False
|
||
assert err is not None
|
||
|
||
def test_validate_strength_no_digit(self):
|
||
"""缺少数字"""
|
||
handler = PasswordHandler()
|
||
ok, err = handler.validate_strength("NoDigitHere")
|
||
assert ok is False
|
||
assert err is not None
|
||
|
||
def test_hash_empty_password(self):
|
||
"""空密码哈希报错"""
|
||
handler = PasswordHandler()
|
||
with pytest.raises((ValueError, Exception)):
|
||
handler.hash_password("")
|
||
|
||
def test_custom_rounds(self):
|
||
"""自定义 rounds(用低轮次测试更快)"""
|
||
handler = PasswordHandler(rounds=4)
|
||
hashed = handler.hash_password("TestPass123")
|
||
assert handler.verify_password("TestPass123", hashed)
|
||
|
||
|
||
# ============================================================
|
||
# PasswordHandler - 全局配置
|
||
# ============================================================
|
||
|
||
|
||
class TestPasswordGlobalConfig:
|
||
"""Password 全局配置与获取"""
|
||
|
||
def test_get_default_handler(self):
|
||
"""未配置时 get 返回默认实例"""
|
||
# 重置默认实例
|
||
with patch("packages.application.auth.password_handler._default_handler", None):
|
||
handler = get_password_handler()
|
||
assert isinstance(handler, PasswordHandler)
|
||
|
||
def test_configure_and_get(self):
|
||
"""配置后可以获取"""
|
||
handler = configure_password_handler(rounds=4)
|
||
assert isinstance(handler, PasswordHandler)
|
||
got = get_password_handler()
|
||
assert got is handler
|
||
|
||
def test_reconfigure_replaces(self):
|
||
"""重新配置会替换"""
|
||
h1 = configure_password_handler(rounds=4)
|
||
h2 = configure_password_handler(rounds=6)
|
||
assert h1 is not h2
|