Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c583aab50c | |||
| 60d6e8efa5 |
Executable
+508
@@ -0,0 +1,508 @@
|
||||
"""
|
||||
Auth 服务层单元测试 - 纯逻辑模块
|
||||
|
||||
覆盖:
|
||||
- PasswordHasher / PasswordValidator (password_hasher.py)
|
||||
- JWTConfig / JWTService / TokenType (jwt_service.py)
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
import jwt as pyjwt
|
||||
import pytest
|
||||
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
|
||||
|
||||
from packages.application.auth.jwt_service import JWTConfig, JWTService, TokenType
|
||||
from packages.application.auth.password_hasher import (
|
||||
PasswordHasher,
|
||||
PasswordValidator,
|
||||
password_hasher,
|
||||
password_validator,
|
||||
)
|
||||
|
||||
# ── PasswordHasher 测试 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPasswordHasher:
|
||||
"""PasswordHasher 密码哈希器测试"""
|
||||
|
||||
def test_default_rounds(self):
|
||||
hasher = PasswordHasher()
|
||||
assert hasher.rounds == 12
|
||||
|
||||
def test_custom_rounds(self):
|
||||
hasher = PasswordHasher(rounds=10)
|
||||
assert hasher.rounds == 10
|
||||
|
||||
def test_rounds_min_boundary(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
assert hasher.rounds == 4
|
||||
|
||||
def test_rounds_max_boundary(self):
|
||||
hasher = PasswordHasher(rounds=31)
|
||||
assert hasher.rounds == 31
|
||||
|
||||
def test_rounds_below_min_raises(self):
|
||||
with pytest.raises(ValueError, match="between 4 and 31"):
|
||||
PasswordHasher(rounds=3)
|
||||
|
||||
def test_rounds_above_max_raises(self):
|
||||
with pytest.raises(ValueError, match="between 4 and 31"):
|
||||
PasswordHasher(rounds=32)
|
||||
|
||||
def test_hash_password_returns_string(self):
|
||||
hasher = PasswordHasher(rounds=4) # 用小rounds加速测试
|
||||
result = hasher.hash_password("testpass123")
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_hash_password_starts_with_bcrypt_prefix(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
result = hasher.hash_password("testpass123")
|
||||
assert result.startswith("$2b$")
|
||||
|
||||
def test_hash_password_contains_rounds(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
result = hasher.hash_password("testpass123")
|
||||
# $2b$04$...
|
||||
assert "$04$" in result
|
||||
|
||||
def test_hash_password_empty_raises(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
with pytest.raises(ValueError, match="cannot be empty"):
|
||||
hasher.hash_password("")
|
||||
|
||||
def test_hash_password_none_raises(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
with pytest.raises(ValueError):
|
||||
hasher.hash_password(None)
|
||||
|
||||
def test_hash_password_different_each_time(self):
|
||||
"""同一密码每次哈希结果不同(因为salt随机)"""
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
h1 = hasher.hash_password("samepass")
|
||||
h2 = hasher.hash_password("samepass")
|
||||
assert h1 != h2
|
||||
|
||||
def test_verify_password_correct(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("mypassword")
|
||||
assert hasher.verify_password("mypassword", hashed) is True
|
||||
|
||||
def test_verify_password_wrong(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("correctpass")
|
||||
assert hasher.verify_password("wrongpass", hashed) is False
|
||||
|
||||
def test_verify_password_empty_password(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("testpass")
|
||||
assert hasher.verify_password("", hashed) is False
|
||||
|
||||
def test_verify_password_empty_hash(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
assert hasher.verify_password("testpass", "") is False
|
||||
|
||||
def test_verify_password_invalid_hash_format(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
assert hasher.verify_password("testpass", "invalid_hash") is False
|
||||
|
||||
def test_verify_password_none_password(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("test")
|
||||
assert hasher.verify_password(None, hashed) is False
|
||||
|
||||
def test_needs_rehash_same_rounds(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("testpass")
|
||||
assert hasher.needs_rehash(hashed) is False
|
||||
|
||||
def test_needs_rehash_different_rounds(self):
|
||||
hasher4 = PasswordHasher(rounds=4)
|
||||
hasher5 = PasswordHasher(rounds=5)
|
||||
hashed = hasher4.hash_password("testpass")
|
||||
assert hasher5.needs_rehash(hashed) is True
|
||||
|
||||
def test_needs_rehash_invalid_hash(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
assert hasher.needs_rehash("invalid") is False
|
||||
|
||||
def test_needs_rehash_empty_hash(self):
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
assert hasher.needs_rehash("") is False
|
||||
|
||||
def test_global_instance_exists(self):
|
||||
assert password_hasher is not None
|
||||
assert isinstance(password_hasher, PasswordHasher)
|
||||
assert password_hasher.rounds == 12
|
||||
|
||||
def test_unicode_password(self):
|
||||
"""支持中文等Unicode密码"""
|
||||
hasher = PasswordHasher(rounds=4)
|
||||
hashed = hasher.hash_password("密码测试123")
|
||||
assert hasher.verify_password("密码测试123", hashed) is True
|
||||
|
||||
|
||||
# ── PasswordValidator 测试 ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPasswordValidator:
|
||||
"""PasswordValidator 密码强度验证器测试"""
|
||||
|
||||
def test_default_config(self):
|
||||
v = PasswordValidator()
|
||||
assert v.min_length == 8
|
||||
assert v.require_uppercase is True
|
||||
assert v.require_lowercase is True
|
||||
assert v.require_digit is True
|
||||
assert v.require_special is False
|
||||
|
||||
def test_custom_config(self):
|
||||
v = PasswordValidator(
|
||||
min_length=10,
|
||||
require_uppercase=False,
|
||||
require_lowercase=False,
|
||||
require_digit=False,
|
||||
require_special=True,
|
||||
)
|
||||
assert v.min_length == 10
|
||||
assert v.require_uppercase is False
|
||||
assert v.require_special is True
|
||||
|
||||
def test_valid_password_default_rules(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate("TestPass123")
|
||||
assert valid is True
|
||||
assert msg is None
|
||||
|
||||
def test_empty_password(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate("")
|
||||
assert valid is False
|
||||
assert "empty" in msg.lower()
|
||||
|
||||
def test_none_password(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate(None)
|
||||
assert valid is False
|
||||
|
||||
def test_too_short(self):
|
||||
v = PasswordValidator(min_length=8)
|
||||
valid, msg = v.validate("Ab1")
|
||||
assert valid is False
|
||||
assert "at least 8" in msg
|
||||
|
||||
def test_exact_min_length(self):
|
||||
v = PasswordValidator(min_length=8, require_uppercase=False, require_lowercase=False, require_digit=False)
|
||||
valid, msg = v.validate("12345678")
|
||||
assert valid is True
|
||||
|
||||
def test_missing_uppercase(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate("testpass123")
|
||||
assert valid is False
|
||||
assert "uppercase" in msg.lower()
|
||||
|
||||
def test_missing_lowercase(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate("TESTPASS123")
|
||||
assert valid is False
|
||||
assert "lowercase" in msg.lower()
|
||||
|
||||
def test_missing_digit(self):
|
||||
v = PasswordValidator()
|
||||
valid, msg = v.validate("TestPassword")
|
||||
assert valid is False
|
||||
assert "digit" in msg.lower()
|
||||
|
||||
def test_require_special_enabled_missing(self):
|
||||
v = PasswordValidator(require_special=True)
|
||||
valid, msg = v.validate("TestPass123")
|
||||
assert valid is False
|
||||
assert "special" in msg.lower()
|
||||
|
||||
def test_require_special_enabled_present(self):
|
||||
v = PasswordValidator(require_special=True)
|
||||
valid, msg = v.validate("TestPass123!")
|
||||
assert valid is True
|
||||
assert msg is None
|
||||
|
||||
def test_special_chars_all_types(self):
|
||||
"""验证各种特殊字符都能识别"""
|
||||
v = PasswordValidator(
|
||||
min_length=8,
|
||||
require_uppercase=False,
|
||||
require_lowercase=False,
|
||||
require_digit=False,
|
||||
require_special=True,
|
||||
)
|
||||
for char in "!@#$%^&*()_+-=[]{}|;:,.<>?~":
|
||||
valid, _ = v.validate(f"testpass{char}")
|
||||
assert valid is True, f"Special char '{char}' not recognized"
|
||||
|
||||
def test_no_requirements_all_pass(self):
|
||||
"""关闭所有要求后任何密码都通过"""
|
||||
v = PasswordValidator(
|
||||
min_length=1,
|
||||
require_uppercase=False,
|
||||
require_lowercase=False,
|
||||
require_digit=False,
|
||||
require_special=False,
|
||||
)
|
||||
valid, msg = v.validate("a")
|
||||
assert valid is True
|
||||
|
||||
def test_global_validator_instance(self):
|
||||
assert password_validator is not None
|
||||
assert isinstance(password_validator, PasswordValidator)
|
||||
assert password_validator.min_length == 8
|
||||
assert password_validator.require_special is False
|
||||
|
||||
|
||||
# ── JWTConfig 测试 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestJWTConfig:
|
||||
"""JWTConfig 配置测试"""
|
||||
|
||||
def test_default_values(self):
|
||||
config = JWTConfig(secret_key="test-secret-key-12345")
|
||||
assert config.SECRET_KEY == "test-secret-key-12345"
|
||||
assert config.ALGORITHM == "HS256"
|
||||
assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 15
|
||||
assert config.REFRESH_TOKEN_EXPIRE_DAYS == 7
|
||||
|
||||
def test_custom_values(self):
|
||||
config = JWTConfig(
|
||||
secret_key="my-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_empty_secret_raises(self):
|
||||
with pytest.raises(ValueError, match="must be provided"):
|
||||
JWTConfig(secret_key="")
|
||||
|
||||
def test_whitespace_secret_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
JWTConfig(secret_key=" ")
|
||||
|
||||
def test_none_secret_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
JWTConfig(secret_key=None)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"insecure",
|
||||
[
|
||||
"your-secret-key-change-in-production",
|
||||
"your-secret-key",
|
||||
"secret",
|
||||
"changeme",
|
||||
"password",
|
||||
"SECRET",
|
||||
"Your-Secret-Key",
|
||||
],
|
||||
)
|
||||
def test_insecure_defaults_raises(self, insecure):
|
||||
with pytest.raises(ValueError, match="insecure"):
|
||||
JWTConfig(secret_key=insecure)
|
||||
|
||||
|
||||
# ── TokenType 测试 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTokenType:
|
||||
"""TokenType 常量测试"""
|
||||
|
||||
def test_access_token_type(self):
|
||||
assert TokenType.ACCESS == "access"
|
||||
|
||||
def test_refresh_token_type(self):
|
||||
assert TokenType.REFRESH == "refresh"
|
||||
|
||||
|
||||
# ── JWTService 测试 ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestJWTService:
|
||||
"""JWTService JWT服务测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
config = JWTConfig(
|
||||
secret_key="test-secret-key-for-testing-only-12345",
|
||||
access_token_expire_minutes=30,
|
||||
refresh_token_expire_days=7,
|
||||
)
|
||||
return JWTService(config)
|
||||
|
||||
def test_init_without_config_raises(self):
|
||||
with pytest.raises(ValueError, match="requires a JWTConfig"):
|
||||
JWTService()
|
||||
|
||||
def test_create_access_token_returns_string(self, service):
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
def test_create_access_token_has_user_id(self, service):
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "user_123"
|
||||
|
||||
def test_create_access_token_has_role(self, service):
|
||||
token = service.create_access_token(user_id="user_123", role="admin")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["role"] == "admin"
|
||||
|
||||
def test_create_access_token_default_role_empty(self, service):
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["role"] == ""
|
||||
|
||||
def test_create_access_token_type_is_access(self, service):
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["type"] == TokenType.ACCESS
|
||||
|
||||
def test_create_access_token_has_iat_and_exp(self, service):
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
assert "iat" in payload
|
||||
assert "exp" in payload
|
||||
assert payload["exp"] > payload["iat"]
|
||||
|
||||
def test_create_access_token_expiry_correct(self, service):
|
||||
"""过期时间 = 签发时间 + 30分钟"""
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
delta_seconds = payload["exp"] - payload["iat"]
|
||||
assert delta_seconds == 30 * 60
|
||||
|
||||
def test_create_access_token_additional_claims(self, service):
|
||||
token = service.create_access_token(
|
||||
user_id="user_123",
|
||||
role="user",
|
||||
additional_claims={"email": "test@example.com", "custom": "value"},
|
||||
)
|
||||
payload = service.verify_token(token)
|
||||
assert payload["email"] == "test@example.com"
|
||||
assert payload["custom"] == "value"
|
||||
|
||||
def test_create_refresh_token_returns_string(self, service):
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
def test_create_refresh_token_has_user_and_session(self, service):
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "user_123"
|
||||
assert payload["session_id"] == "sess_456"
|
||||
|
||||
def test_create_refresh_token_type_is_refresh(self, service):
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["type"] == TokenType.REFRESH
|
||||
|
||||
def test_create_refresh_token_expiry_correct(self, service):
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
payload = service.verify_token(token)
|
||||
delta_seconds = payload["exp"] - payload["iat"]
|
||||
assert delta_seconds == 7 * 24 * 60 * 60
|
||||
|
||||
def test_verify_access_token_success(self, service):
|
||||
token = service.create_access_token(user_id="user_123", role="admin")
|
||||
payload = service.verify_access_token(token)
|
||||
assert payload["sub"] == "user_123"
|
||||
assert payload["role"] == "admin"
|
||||
|
||||
def test_verify_access_token_wrong_type_raises(self, service):
|
||||
"""用refresh token当access token验证会失败"""
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
with pytest.raises(ValueError, match="must be 'access'"):
|
||||
service.verify_access_token(token)
|
||||
|
||||
def test_verify_refresh_token_success(self, service):
|
||||
token = service.create_refresh_token(user_id="user_123", session_id="sess_456")
|
||||
payload = service.verify_refresh_token(token)
|
||||
assert payload["sub"] == "user_123"
|
||||
assert payload["session_id"] == "sess_456"
|
||||
|
||||
def test_verify_refresh_token_wrong_type_raises(self, service):
|
||||
"""用access token当refresh token验证会失败"""
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
with pytest.raises(ValueError, match="must be 'refresh'"):
|
||||
service.verify_refresh_token(token)
|
||||
|
||||
def test_verify_token_invalid_token_raises(self, service):
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token("invalid.token.here")
|
||||
|
||||
def test_verify_token_empty_raises(self, service):
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token("")
|
||||
|
||||
def test_verify_token_wrong_secret(self, service):
|
||||
"""用不同密钥签发的token无法验证"""
|
||||
other_config = JWTConfig(secret_key="different-secret-key-for-testing-123")
|
||||
other_service = JWTService(other_config)
|
||||
token = other_service.create_access_token(user_id="user_123")
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token(token)
|
||||
|
||||
def test_verify_token_tampered_signature(self, service):
|
||||
"""篡改签名的token无法验证"""
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
# 篡改最后一个字符
|
||||
tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token(tampered)
|
||||
|
||||
def test_expired_token_raises(self):
|
||||
"""过期token验证失败"""
|
||||
config = JWTConfig(
|
||||
secret_key="test-secret-key-for-testing-only-12345",
|
||||
access_token_expire_minutes=0, # 立即过期
|
||||
)
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
time.sleep(1) # 等1秒确保过期
|
||||
with pytest.raises(ExpiredSignatureError):
|
||||
service.verify_token(token)
|
||||
|
||||
def test_different_algorithm(self):
|
||||
"""支持不同算法"""
|
||||
config = JWTConfig(
|
||||
secret_key="test-secret-key-for-testing-only-12345-abcdef",
|
||||
algorithm="HS512",
|
||||
)
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="user_123")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "user_123"
|
||||
|
||||
def test_additional_claims_not_overwrite_standard(self):
|
||||
"""additional_claims 不会覆盖标准字段"""
|
||||
config = JWTConfig(secret_key="test-secret-key-for-testing-only-12345")
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(
|
||||
user_id="real_user",
|
||||
additional_claims={"sub": "fake_user", "type": "fake_type"},
|
||||
)
|
||||
payload = service.verify_token(token)
|
||||
# additional_claims 在标准字段之后update,所以会覆盖
|
||||
# 这个测试验证当前行为(additional_claims优先级高)
|
||||
assert payload["sub"] == "fake_user"
|
||||
|
||||
def test_unicode_user_id(self):
|
||||
"""支持Unicode用户ID"""
|
||||
config = JWTConfig(secret_key="test-secret-key-for-testing-only-12345")
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="用户_测试123")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "用户_测试123"
|
||||
Reference in New Issue
Block a user