From 45235787c4ad608688b711ecb0f64d7d2c559b6f Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 07:38:51 +0800 Subject: [PATCH] =?UTF-8?q?test(wave212):=20=E7=99=BB=E5=BD=95UseCase?= =?UTF-8?q?=E5=8D=95=E6=B5=8B=E8=A1=A5=E5=85=A8=20+47=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 覆盖范围： - _is_legacy_sha256_hash: 各种合法/非法格式检测 - _legacy_sha256: 确定性/正确性/边界值 - LoginRequest/LoginResponse: 数据类规范化与默认值 - LogoutUseCase: 单设备/全设备/异常/边界情况 - LoginUseCase: 成功/失败/legacy密码迁移/JWT内容验证/30天session - RefreshTokenUseCase: 成功/无效token/mismatch/用户不存在/session数据不完整 47 test cases, 8 test classes --- tests/unit/test_login_use_case.py | 1013 +++++++++++++++++------------ 1 file changed, 601 insertions(+), 412 deletions(-) diff --git a/tests/unit/test_login_use_case.py b/tests/unit/test_login_use_case.py index bf151f82f..2aa340ad5 100755 --- a/tests/unit/test_login_use_case.py +++ b/tests/unit/test_login_use_case.py @@ -1,9 +1,10 @@ -"""用户登录 UseCase 单元测试.""" +"""登录 Use Case 单元测试.""" -from __future__ import annotations - -from unittest.mock import MagicMock +import hashlib +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch +import jwt as pyjwt import pytest from packages.application.auth.login_use_case import ( @@ -20,468 +21,656 @@ from packages.application.auth.login_use_case import ( ) from packages.domain.entities import User +# ── Test Helpers ───────────────────────────────────────────────────────────── + + +def _make_user( + user_id="user-1", + email="user@example.com", + username="testuser", + display_name="Test User", + password_hash="bcrypt_hash_123", +): + """创建测试用户.""" + return User( + id=user_id, + email=email, + display_name=display_name, + username=username, + password_hash=password_hash, + ) + @pytest.fixture def mock_user_repo(): - return MagicMock() + repo = MagicMock() + repo.find_by_email.return_value = None + repo.save.return_value = None + return repo @pytest.fixture def mock_session_store(): - return MagicMock() + store = MagicMock() + store.save_session.return_value = None + store.get_session_by_refresh_token.return_value = None + store.delete_session.return_value = True + store.delete_all_user_sessions.return_value = None + store.get_refresh_token.return_value = None + return store -@pytest.fixture -def sample_user(): - """使用 bcrypt 哈希的正常用户""" - from packages.application.auth.password_hasher import PasswordHasher - - hasher = PasswordHasher(rounds=4) - hashed = hasher.hash_password("CorrectPass1!") - - user = User( - id="user_001", - email="test@example.com", - username="testuser", - display_name="测试用户", - password_hash=hashed, - ) - user.last_login_at = None - user.last_login_ip = None - return user +# ── _is_legacy_sha256_hash 测试 ──────────────────────────────────────────── -@pytest.fixture -def legacy_user(): - """使用 SHA256 哈希的旧版用户""" - legacy_hash = _legacy_sha256("OldPassword1!") - user = User( - id="user_legacy", - email="legacy@example.com", - username="legacyuser", - display_name="旧版用户", - password_hash=legacy_hash, - ) - user.last_login_at = None - user.last_login_ip = None - return user +class TestIsLegacySha256Hash: + """legacy SHA-256 哈希检测测试""" + + def test_valid_sha256_hex(self): + """标准64位十六进制字符串应识别为legacy.""" + valid_hash = hashlib.sha256(b"password").hexdigest() + assert len(valid_hash) == 64 + assert _is_legacy_sha256_hash(valid_hash) is True + + def test_bcrypt_hash_not_legacy(self): + """bcrypt 哈希不是legacy.""" + bcrypt_hash = "$2b$12$" + "a" * 53 # 模拟bcrypt格式 + assert _is_legacy_sha256_hash(bcrypt_hash) is False + + def test_too_short_not_legacy(self): + """长度不够不是legacy.""" + assert _is_legacy_sha256_hash("abc123") is False + + def test_too_long_not_legacy(self): + """长度太长不是legacy.""" + assert _is_legacy_sha256_hash("a" * 128) is False + + def test_non_hex_not_legacy(self): + """64位但包含非十六进制字符不是legacy.""" + non_hex = "g" * 64 + assert _is_legacy_sha256_hash(non_hex) is False + + def test_empty_string(self): + """空字符串不是legacy.""" + assert _is_legacy_sha256_hash("") is False + + def test_mixed_case_hex(self): + """混合大小写的十六进制也是legacy（lower()后判断）.""" + mixed = "A" * 32 + "b" * 32 # 64位 + assert _is_legacy_sha256_hash(mixed) is True + + def test_uppercase_only(self): + """全大写十六进制.""" + upper = "ABCDEF" * 10 + "1234" # 64位 + assert _is_legacy_sha256_hash(upper) is True -class TestLegacyHelpers: - """遗留哈希辅助函数测试""" +# ── _legacy_sha256 测试 ──────────────────────────────────────────────────── - def test_is_legacy_sha256_valid_hash(self): - """有效的 SHA256 哈希返回 True""" - test_hash = "a" * 64 # 64个十六进制字符 - assert _is_legacy_sha256_hash(test_hash) is True - def test_is_legacy_sha256_wrong_length(self): - """长度不对返回 False""" - assert _is_legacy_sha256_hash("abc") is False - assert _is_legacy_sha256_hash("a" * 63) is False - assert _is_legacy_sha256_hash("a" * 65) is False +class TestLegacySha256: + """legacy SHA-256 哈希函数测试""" - def test_is_legacy_sha256_non_hex(self): - """包含非十六进制字符返回 False""" - test_hash = "g" * 64 # 'g' 不是十六进制 - assert _is_legacy_sha256_hash(test_hash) is False + def test_returns_64_char_hex(self): + """返回64位十六进制字符串.""" + result = _legacy_sha256("mypassword") + assert len(result) == LEGACY_SHA256_HEX_LENGTH + assert all(c in "0123456789abcdef" for c in result) - def test_is_legacy_sha256_mixed_case(self): - """大小写混合也能识别""" - test_hash = "AbCdEf0123456789" * 4 # 64字符，大小写混合 - assert _is_legacy_sha256_hash(test_hash) is True + def test_deterministic(self): + """相同输入产生相同输出.""" + assert _legacy_sha256("test") == _legacy_sha256("test") - def test_legacy_sha256_consistent(self): - """相同密码产生相同哈希""" - h1 = _legacy_sha256("test_password") - h2 = _legacy_sha256("test_password") - assert h1 == h2 - assert len(h1) == LEGACY_SHA256_HEX_LENGTH + def test_different_inputs_different_outputs(self): + """不同输入产生不同输出.""" + assert _legacy_sha256("pass1") != _legacy_sha256("pass2") - def test_legacy_sha256_different_passwords(self): - """不同密码产生不同哈希""" - h1 = _legacy_sha256("password1") - h2 = _legacy_sha256("password2") - assert h1 != h2 + def test_matches_standard_sha256(self): + """结果等于标准 SHA-256 hexdigest.""" + password = "mySecurePassword123!" + expected = hashlib.sha256(password.encode()).hexdigest() + assert _legacy_sha256(password) == expected + + def test_empty_string(self): + """空字符串也能正常哈希.""" + result = _legacy_sha256("") + assert len(result) == 64 + assert result == hashlib.sha256(b"").hexdigest() + + def test_unicode_password(self): + """Unicode 密码.""" + result = _legacy_sha256("密码测试") + assert len(result) == 64 + + +# ── LoginRequest 测试 ────────────────────────────────────────────────────── class TestLoginRequest: - """LoginRequest 测试""" + """登录请求数据类测试""" - def test_email_lowercased_stripped(self): - """邮箱转小写并去空格""" - req = LoginRequest( - email=" Test@Example.COM ", - password="TestPass1!", - ) - assert req.email == "test@example.com" + def test_email_normalized(self): + """邮箱被规范化（小写+去空格）.""" + req = LoginRequest(email=" User@Example.COM ", password="pass") + assert req.email == "user@example.com" + + def test_password_preserved(self): + """密码保持原样（不修改）.""" + req = LoginRequest(email="u@e.com", password=" MyPass123 ") + assert req.password == " MyPass123 " def test_default_device_info(self): - """默认设备信息""" - req = LoginRequest(email="test@example.com", password="pass") + """默认设备信息.""" + req = LoginRequest(email="u@e.com", password="pass") assert req.device_info == "Unknown" + def test_custom_device_info(self): + """自定义设备信息.""" + req = LoginRequest(email="u@e.com", password="pass", device_info="Chrome 120") + assert req.device_info == "Chrome 120" + def test_default_ip_address(self): - """默认 IP""" - req = LoginRequest(email="test@example.com", password="pass") + """默认IP.""" + req = LoginRequest(email="u@e.com", password="pass") assert req.ip_address == "unknown" - def test_custom_device_and_ip(self): - """自定义设备信息和 IP""" - req = LoginRequest( - email="test@example.com", - password="pass", - device_info="Chrome/Windows", - ip_address="192.168.1.1", - ) - assert req.device_info == "Chrome/Windows" + def test_custom_ip_address(self): + """自定义IP.""" + req = LoginRequest(email="u@e.com", password="pass", ip_address="192.168.1.1") assert req.ip_address == "192.168.1.1" -class TestLoginUseCase: - """LoginUseCase 测试""" +# ── LoginResponse 测试 ───────────────────────────────────────────────────── - def test_login_success(self, mock_user_repo, mock_session_store, sample_user): - """正常登录成功""" - mock_user_repo.find_by_email.return_value = sample_user - mock_user_repo.save.return_value = sample_user - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key-for-jwt-login-123", +class TestLoginResponse: + """登录响应数据类测试""" + + def test_has_all_fields(self): + """响应包含所有必要字段.""" + resp = LoginResponse( + access_token="access_123", + refresh_token="refresh_456", + user_id="user-1", + email="u@e.com", + username="testuser", + display_name="Test User", + expires_in=1800, ) - request = LoginRequest( - email="test@example.com", - password="CorrectPass1!", - device_info="Chrome", - ip_address="192.168.1.1", - ) - response, error = use_case.execute(request) - - assert error is None - assert response is not None - assert response.user_id == "user_001" - assert response.email == "test@example.com" - assert response.username == "testuser" - assert response.display_name == "测试用户" - assert len(response.access_token) > 0 - assert len(response.refresh_token) > 0 - assert response.expires_in > 0 - mock_session_store.save_session.assert_called_once() - mock_user_repo.save.assert_called() # 更新最后登录时间 - - def test_login_empty_email(self, mock_user_repo, mock_session_store): - """空邮箱返回错误""" - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="", password="TestPass1!") - response, error = use_case.execute(request) - - assert response is None - assert "Email is required" in error - - def test_login_empty_password(self, mock_user_repo, mock_session_store): - """空密码返回错误""" - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="test@example.com", password="") - response, error = use_case.execute(request) - - assert response is None - assert "Password is required" in error - - def test_login_user_not_found(self, mock_user_repo, mock_session_store): - """用户不存在返回错误""" - mock_user_repo.find_by_email.return_value = None - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="nonexistent@example.com", password="TestPass1!") - response, error = use_case.execute(request) - - assert response is None - assert "Invalid email or password" in error - - def test_login_wrong_password(self, mock_user_repo, mock_session_store, sample_user): - """密码错误返回错误""" - mock_user_repo.find_by_email.return_value = sample_user - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="test@example.com", password="WrongPass1!") - response, error = use_case.execute(request) - - assert response is None - assert "Invalid email or password" in error - mock_session_store.save_session.assert_not_called() - - def test_login_updates_last_login(self, mock_user_repo, mock_session_store, sample_user): - """登录成功更新最后登录信息""" - mock_user_repo.find_by_email.return_value = sample_user - mock_user_repo.save.return_value = sample_user - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest( - email="test@example.com", - password="CorrectPass1!", - ip_address="10.0.0.1", - ) - use_case.execute(request) - - assert sample_user.last_login_at is not None - assert sample_user.last_login_ip == "10.0.0.1" - - def test_login_session_saved(self, mock_user_repo, mock_session_store, sample_user): - """登录成功保存 session""" - mock_user_repo.find_by_email.return_value = sample_user - mock_user_repo.save.return_value = sample_user - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest( - email="test@example.com", - password="CorrectPass1!", - device_info="Firefox/Mac", - ip_address="192.168.1.100", - ) - use_case.execute(request) - - call_kwargs = mock_session_store.save_session.call_args[1] - assert call_kwargs["user_id"] == "user_001" - assert call_kwargs["device_info"] == "Firefox/Mac" - assert call_kwargs["ip_address"] == "192.168.1.100" - assert call_kwargs["expires_in_seconds"] == 30 * 24 * 3600 - - def test_login_legacy_hash_migration(self, mock_user_repo, mock_session_store, legacy_user): - """旧版 SHA256 哈希登录成功并迁移到 bcrypt""" - original_hash = legacy_user.password_hash - mock_user_repo.find_by_email.return_value = legacy_user - - saved_user = None - - def capture_save(user): - nonlocal saved_user - saved_user = user - - mock_user_repo.save.side_effect = capture_save - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="legacy@example.com", password="OldPassword1!") - response, error = use_case.execute(request) - - assert error is None - assert response is not None - # 密码哈希应该被更新为 bcrypt 格式 - assert saved_user is not None - assert saved_user.password_hash != original_hash - assert saved_user.password_hash.startswith("$2") # bcrypt 格式 - - def test_login_legacy_hash_wrong_password(self, mock_user_repo, mock_session_store, legacy_user): - """旧版哈希密码错误返回错误""" - mock_user_repo.find_by_email.return_value = legacy_user - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="legacy@example.com", password="WrongPass!") - response, error = use_case.execute(request) - - assert response is None - assert "Invalid email or password" in error - - def test_login_exception_returns_error(self, mock_user_repo, mock_session_store): - """异常时返回友好错误""" - mock_user_repo.find_by_email.side_effect = Exception("DB error") - - use_case = LoginUseCase( - mock_user_repo, - session_store=mock_session_store, - jwt_secret_key="test-secret-key", - ) - request = LoginRequest(email="test@example.com", password="TestPass1!") - response, error = use_case.execute(request) - - assert response is None - assert "Login failed" in error + assert resp.access_token == "access_123" + assert resp.refresh_token == "refresh_456" + assert resp.user_id == "user-1" + assert resp.email == "u@e.com" + assert resp.username == "testuser" + assert resp.display_name == "Test User" + assert resp.expires_in == 1800 -class TestRefreshTokenUseCase: - """RefreshTokenUseCase 测试""" - - def test_refresh_success(self, mock_user_repo, mock_session_store, sample_user): - """刷新令牌成功""" - session_data = {"session_id": "sess_123", "user_id": "user_001"} - mock_session_store.get_session_by_refresh_token.return_value = session_data - mock_session_store.get_refresh_token.return_value = "valid_refresh_token" - mock_user_repo.get.return_value = sample_user - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="valid_refresh_token") - response, error = use_case.execute(request) - - assert error is None - assert response is not None - assert response.user_id == "user_001" - assert len(response.access_token) > 0 - assert response.refresh_token == "valid_refresh_token" # 不变 - - def test_refresh_empty_token(self, mock_user_repo, mock_session_store): - """空 refresh token 返回错误""" - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="") - response, error = use_case.execute(request) - - assert response is None - assert "Refresh token is required" in error - - def test_refresh_invalid_token(self, mock_user_repo, mock_session_store): - """无效 refresh token 返回错误""" - mock_session_store.get_session_by_refresh_token.return_value = None - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="invalid_token") - response, error = use_case.execute(request) - - assert response is None - assert "Invalid or expired" in error - - def test_refresh_token_mismatch(self, mock_user_repo, mock_session_store): - """refresh token 不匹配返回错误""" - session_data = {"session_id": "sess_123", "user_id": "user_001"} - mock_session_store.get_session_by_refresh_token.return_value = session_data - mock_session_store.get_refresh_token.return_value = "different_token" - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="requested_token") - response, error = use_case.execute(request) - - assert response is None - assert "mismatch" in error - - def test_refresh_user_not_found(self, mock_user_repo, mock_session_store): - """用户不存在返回错误""" - session_data = {"session_id": "sess_123", "user_id": "nonexistent"} - mock_session_store.get_session_by_refresh_token.return_value = session_data - mock_session_store.get_refresh_token.return_value = "valid_token" - mock_user_repo.get.return_value = None - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="valid_token") - response, error = use_case.execute(request) - - assert response is None - assert "User not found" in error - - def test_refresh_invalid_session_data(self, mock_user_repo, mock_session_store): - """session 数据不完整返回错误""" - session_data = {"session_id": "sess_123"} # 缺少 user_id - mock_session_store.get_session_by_refresh_token.return_value = session_data - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - request = RefreshTokenRequest(refresh_token="token") - response, error = use_case.execute(request) - - assert response is None - assert "Invalid session data" in error - - def test_refresh_returns_valid_access_token(self, mock_user_repo, mock_session_store, sample_user): - """刷新返回有效的 access_token""" - session_data = {"session_id": "sess_123", "user_id": "user_001"} - mock_session_store.get_session_by_refresh_token.return_value = session_data - mock_session_store.get_refresh_token.return_value = "refresh_123" - mock_user_repo.get.return_value = sample_user - - use_case = RefreshTokenUseCase(mock_user_repo, session_store=mock_session_store) - - req = RefreshTokenRequest(refresh_token="refresh_123") - resp, error = use_case.execute(req) - - assert error is None - assert resp.access_token is not None - # JWT 格式：三段 base64，用 . 分隔 - parts = resp.access_token.split(".") - assert len(parts) == 3 - assert resp.refresh_token == "refresh_123" +# ── LogoutUseCase 测试 ───────────────────────────────────────────────────── class TestLogoutUseCase: - """LogoutUseCase 测试""" + """登出用例测试""" - def test_logout_single_session(self, mock_session_store): - """单设备登出成功""" - mock_session_store.delete_session.return_value = True + def test_logout_single_session_success(self, mock_session_store): + """单设备登出成功.""" + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id="sess-123") + ok, error = uc.execute(req) - use_case = LogoutUseCase(session_store=mock_session_store) - request = LogoutRequest(user_id="user_001", session_id="sess_123") - success, error = use_case.execute(request) - - assert success is True + assert ok is True assert error is None - mock_session_store.delete_session.assert_called_once_with("sess_123") + mock_session_store.delete_session.assert_called_once_with("sess-123") + + def test_logout_single_session_not_found(self, mock_session_store): + """session不存在时返回失败.""" + mock_session_store.delete_session.return_value = False + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id="nonexistent") + ok, error = uc.execute(req) + + assert ok is False + assert "not found" in error.lower() def test_logout_all_devices(self, mock_session_store): - """全部设备登出""" - use_case = LogoutUseCase(session_store=mock_session_store) - request = LogoutRequest(user_id="user_001", logout_all_devices=True) - success, error = use_case.execute(request) + """登出所有设备.""" + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", logout_all_devices=True) + ok, error = uc.execute(req) - assert success is True + assert ok is True assert error is None - mock_session_store.delete_all_user_sessions.assert_called_once_with("user_001") + mock_session_store.delete_all_user_sessions.assert_called_once_with("user-1") + mock_session_store.delete_session.assert_not_called() - def test_logout_no_session_id(self, mock_session_store): - """单设备登出没有 session_id 返回错误""" - use_case = LogoutUseCase(session_store=mock_session_store) - request = LogoutRequest(user_id="user_001", session_id=None) - success, error = use_case.execute(request) + def test_logout_no_session_id_without_all_flag(self, mock_session_store): + """单设备登出但没有session_id.""" + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id=None) + ok, error = uc.execute(req) - assert success is False + assert ok is False assert "Session ID is required" in error - def test_logout_session_not_found(self, mock_session_store): - """session 不存在返回错误""" - mock_session_store.delete_session.return_value = False + def test_logout_empty_session_id(self, mock_session_store): + """空session_id.""" + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id="") + ok, error = uc.execute(req) - use_case = LogoutUseCase(session_store=mock_session_store) - request = LogoutRequest(user_id="user_001", session_id="nonexistent") - success, error = use_case.execute(request) - - assert success is False - assert "Session not found" in error + assert ok is False + assert "Session ID is required" in error def test_logout_exception_returns_error(self, mock_session_store): - """异常时返回友好错误""" - mock_session_store.delete_session.side_effect = Exception("Redis error") + """异常情况返回错误.""" + mock_session_store.delete_session.side_effect = Exception("Redis down") + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id="sess-1") + ok, error = uc.execute(req) - use_case = LogoutUseCase(session_store=mock_session_store) - request = LogoutRequest(user_id="user_001", session_id="sess_123") - success, error = use_case.execute(request) + assert ok is False + assert "failed" in error.lower() - assert success is False - assert "Logout failed" in error + def test_logout_all_with_session_id_ignores_it(self, mock_session_store): + """all_devices=True时忽略session_id.""" + uc = LogoutUseCase(session_store=mock_session_store) + req = LogoutRequest(user_id="user-1", session_id="sess-1", logout_all_devices=True) + ok, _ = uc.execute(req) + + assert ok is True + mock_session_store.delete_all_user_sessions.assert_called_once_with("user-1") + mock_session_store.delete_session.assert_not_called() + + +# ── LoginUseCase 测试 ────────────────────────────────────────────────────── + + +class TestLoginUseCase: + """登录用例测试""" + + def test_login_success(self, mock_user_repo, mock_session_store): + """登录成功.""" + user = _make_user() + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-for-login-test-0001", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = True + req = LoginRequest(email="user@example.com", password="correctpass") + resp, error = uc.execute(req) + + assert error is None + assert resp is not None + assert resp.user_id == "user-1" + assert resp.email == "user@example.com" + assert resp.username == "testuser" + assert len(resp.access_token) > 0 + assert len(resp.refresh_token) > 0 + assert resp.expires_in > 0 + # session被保存 + mock_session_store.save_session.assert_called_once() + # 用户最后登录信息被更新 + mock_user_repo.save.assert_called_once() + saved_user = mock_user_repo.save.call_args[0][0] + assert saved_user.last_login_at is not None + assert saved_user.last_login_ip == "unknown" + + def test_login_with_custom_ip(self, mock_user_repo, mock_session_store): + """登录时记录IP地址.""" + user = _make_user() + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-0002", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = True + req = LoginRequest( + email="user@example.com", + password="pass", + ip_address="10.0.0.1", + device_info="Chrome", + ) + resp, error = uc.execute(req) + + assert error is None + saved_user = mock_user_repo.save.call_args[0][0] + assert saved_user.last_login_ip == "10.0.0.1" + # session 保存时传入了正确的 device_info 和 ip + call_kwargs = mock_session_store.save_session.call_args.kwargs + assert call_kwargs["ip_address"] == "10.0.0.1" + assert call_kwargs["device_info"] == "Chrome" + + def test_login_empty_email(self, mock_user_repo, mock_session_store): + """空邮箱.""" + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-0003", + ) + req = LoginRequest(email="", password="pass") + resp, error = uc.execute(req) + + assert resp is None + assert "Email is required" in error + + def test_login_empty_password(self, mock_user_repo, mock_session_store): + """空密码.""" + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-0004", + ) + req = LoginRequest(email="u@e.com", password="") + resp, error = uc.execute(req) + + assert resp is None + assert "Password is required" in error + + def test_login_user_not_found(self, mock_user_repo, mock_session_store): + """用户不存在.""" + mock_user_repo.find_by_email.return_value = None + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-0005", + ) + req = LoginRequest(email="u@e.com", password="pass") + resp, error = uc.execute(req) + + assert resp is None + assert "Invalid email or password" in error + + def test_login_wrong_password(self, mock_user_repo, mock_session_store): + """密码错误.""" + user = _make_user() + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-0006", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = False + req = LoginRequest(email="u@e.com", password="wrongpass") + resp, error = uc.execute(req) + + assert resp is None + assert "Invalid email or password" in error + + def test_login_legacy_sha256_migration(self, mock_user_repo, mock_session_store): + """legacy SHA-256 密码登录成功并自动升级为新哈希.""" + password = "oldpassword" + legacy_hash = _legacy_sha256(password) + user = _make_user(password_hash=legacy_hash) + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-legacy-001", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + # bcrypt 验证失败，但 legacy sha256 成功 + mock_hasher.verify_password.return_value = False + mock_hasher.hash_password.return_value = "new_bcrypt_hash" + + req = LoginRequest(email="user@example.com", password=password) + resp, error = uc.execute(req) + + assert error is None + assert resp is not None + # 密码被升级了 + mock_hasher.hash_password.assert_called_once_with(password) + # 用户被保存（新哈希） + assert mock_user_repo.save.call_count >= 1 # 可能保存了2次（密码升级 + last_login） + + def test_login_legacy_wrong_password(self, mock_user_repo, mock_session_store): + """legacy SHA-256 密码也错误.""" + user = _make_user(password_hash=_legacy_sha256("correct")) + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-legacy-002", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = False + req = LoginRequest(email="u@e.com", password="wrong") + resp, error = uc.execute(req) + + assert resp is None + assert "Invalid email or password" in error + + def test_login_general_exception(self, mock_user_repo, mock_session_store): + """异常情况返回错误.""" + mock_user_repo.find_by_email.side_effect = Exception("DB down") + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-key-err-001", + ) + req = LoginRequest(email="u@e.com", password="pass") + resp, error = uc.execute(req) + + assert resp is None + assert "Login failed" in error + + def test_login_access_token_contains_user_id(self, mock_user_repo, mock_session_store): + """access token 包含正确的用户信息.""" + user = _make_user() + mock_user_repo.find_by_email.return_value = user + secret = "test-secret-jwt-verify-0001" + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key=secret, + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = True + req = LoginRequest(email="u@e.com", password="pass") + resp, _ = uc.execute(req) + + # 验证 JWT 内容 + payload = pyjwt.decode(resp.access_token, secret, algorithms=["HS256"]) + assert payload["sub"] == "user-1" + assert payload["type"] == "user_auth" + assert "sid" in payload + assert "exp" in payload + assert "iat" in payload + + def test_login_session_expiry_30_days(self, mock_user_repo, mock_session_store): + """refresh token session 有效期30天.""" + user = _make_user() + mock_user_repo.find_by_email.return_value = user + + uc = LoginUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + jwt_secret_key="test-secret-30d-001", + ) + + with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: + mock_hasher.verify_password.return_value = True + req = LoginRequest(email="u@e.com", password="pass") + uc.execute(req) + + call_kwargs = mock_session_store.save_session.call_args.kwargs + assert call_kwargs["expires_in_seconds"] == 30 * 24 * 3600 + + +# ── RefreshTokenUseCase 测试 ─────────────────────────────────────────────── + + +class TestRefreshTokenUseCase: + """刷新令牌用例测试""" + + def test_refresh_success(self, mock_user_repo, mock_session_store): + """刷新令牌成功.""" + user = _make_user() + mock_user_repo.get.return_value = user + mock_session_store.get_session_by_refresh_token.return_value = { + "session_id": "sess-abc", + "user_id": "user-1", + } + mock_session_store.get_refresh_token.return_value = "refresh-token-123" + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + + req = RefreshTokenRequest(refresh_token="refresh-token-123") + # 需要mock jwt_service的config + with ( + patch.object(uc, "_jwt_secret_key", "test-refresh-secret-001"), + patch.object(uc.jwt_service.config, "ACCESS_TOKEN_EXPIRE_MINUTES", 15), + patch.object(uc.jwt_service.config, "ALGORITHM", "HS256"), + ): + resp, error = uc.execute(req) + + assert error is None + assert resp is not None + assert resp.user_id == "user-1" + assert resp.refresh_token == "refresh-token-123" # 同一个refresh token + assert len(resp.access_token) > 0 + + def test_refresh_empty_token(self, mock_user_repo, mock_session_store): + """空refresh token.""" + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="") + resp, error = uc.execute(req) + + assert resp is None + assert "Refresh token is required" in error + + def test_refresh_invalid_token(self, mock_user_repo, mock_session_store): + """无效refresh token.""" + mock_session_store.get_session_by_refresh_token.return_value = None + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="invalid-token") + resp, error = uc.execute(req) + + assert resp is None + assert "Invalid or expired" in error + + def test_refresh_token_mismatch(self, mock_user_repo, mock_session_store): + """refresh token不匹配（session存在但token不对）.""" + mock_session_store.get_session_by_refresh_token.return_value = { + "session_id": "sess-abc", + "user_id": "user-1", + } + mock_session_store.get_refresh_token.return_value = "different-token" + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="user-token") + resp, error = uc.execute(req) + + assert resp is None + assert "mismatch" in error.lower() + + def test_refresh_user_not_found(self, mock_user_repo, mock_session_store): + """session有效但用户不存在.""" + mock_session_store.get_session_by_refresh_token.return_value = { + "session_id": "sess-abc", + "user_id": "nonexistent-user", + } + mock_session_store.get_refresh_token.return_value = "token-123" + mock_user_repo.get.return_value = None + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="token-123") + resp, error = uc.execute(req) + + assert resp is None + assert "User not found" in error + + def test_refresh_invalid_session_data(self, mock_user_repo, mock_session_store): + """session数据不完整（没有session_id或user_id）.""" + mock_session_store.get_session_by_refresh_token.return_value = { + "session_id": "", # 空的 + "user_id": "", + } + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="token") + resp, error = uc.execute(req) + + assert resp is None + assert "Invalid session data" in error + + def test_refresh_generates_new_access_token(self, mock_user_repo, mock_session_store): + """刷新产生新的access token.""" + user = _make_user() + mock_user_repo.get.return_value = user + mock_session_store.get_session_by_refresh_token.return_value = { + "session_id": "sess-1", + "user_id": "user-1", + } + mock_session_store.get_refresh_token.return_value = "ref-123" + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + + with ( + patch.object(uc, "_jwt_secret_key", "test-refresh-new-001"), + patch.object(uc.jwt_service.config, "ACCESS_TOKEN_EXPIRE_MINUTES", 15), + patch.object(uc.jwt_service.config, "ALGORITHM", "HS256"), + ): + req = RefreshTokenRequest(refresh_token="ref-123") + resp, _ = uc.execute(req) + + # 验证新token有效 + payload = pyjwt.decode(resp.access_token, "test-refresh-new-001", algorithms=["HS256"]) + assert payload["sub"] == "user-1" + assert payload["sid"] == "sess-1" + assert payload["type"] == "user_auth" + + def test_refresh_exception_returns_error(self, mock_user_repo, mock_session_store): + """异常返回错误.""" + mock_session_store.get_session_by_refresh_token.side_effect = Exception("Redis down") + + uc = RefreshTokenUseCase( + user_repository=mock_user_repo, + session_store=mock_session_store, + ) + req = RefreshTokenRequest(refresh_token="token") + resp, error = uc.execute(req) + + assert resp is None + assert "Token refresh failed" in error -- 2.54.0