diff --git a/tests/unit/test_login_use_case.py b/tests/unit/test_login_use_case.py index e9ce184c7..bf151f82f 100755 --- a/tests/unit/test_login_use_case.py +++ b/tests/unit/test_login_use_case.py @@ -1,13 +1,15 @@ -""" -登录/登出/刷新令牌 Use Case 测试 -""" +"""用户登录 UseCase 单元测试.""" -from unittest.mock import Mock, patch +from __future__ import annotations + +from unittest.mock import MagicMock import pytest from packages.application.auth.login_use_case import ( + LEGACY_SHA256_HEX_LENGTH, LoginRequest, + LoginResponse, LoginUseCase, LogoutRequest, LogoutUseCase, @@ -19,406 +21,467 @@ from packages.application.auth.login_use_case import ( from packages.domain.entities import User -class TestLegacyHashHelpers: - """旧版密码哈希工具函数测试""" +@pytest.fixture +def mock_user_repo(): + return MagicMock() - def test_is_legacy_sha256_hash_valid(self): - """测试识别有效的 SHA256 哈希""" - valid_hash = "a" * 64 # 64个十六进制字符 - assert _is_legacy_sha256_hash(valid_hash) is True - def test_is_legacy_sha256_hash_wrong_length(self): - """测试长度不对的不是 SHA256""" - assert _is_legacy_sha256_hash("abc123") is False +@pytest.fixture +def mock_session_store(): + return MagicMock() + + +@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 + + +@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 TestLegacyHelpers: + """遗留哈希辅助函数测试""" + + 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 - def test_is_legacy_sha256_hash_non_hex(self): - """测试包含非十六进制字符的不是 SHA256""" - non_hex = "g" * 64 - assert _is_legacy_sha256_hash(non_hex) is False + 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_legacy_sha256_produces_correct_hash(self): - """测试 SHA256 哈希生成正确""" - result = _legacy_sha256("password123") - assert len(result) == 64 - assert all(c in "0123456789abcdef" for c in result) - # 相同输入产生相同输出 - assert _legacy_sha256("password123") == 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_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_legacy_sha256_different_passwords(self): + """不同密码产生不同哈希""" + h1 = _legacy_sha256("password1") + h2 = _legacy_sha256("password2") + assert h1 != h2 + + +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_default_device_info(self): + """默认设备信息""" + req = LoginRequest(email="test@example.com", password="pass") + assert req.device_info == "Unknown" + + def test_default_ip_address(self): + """默认 IP""" + req = LoginRequest(email="test@example.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" + assert req.ip_address == "192.168.1.1" class TestLoginUseCase: - """登录用例测试""" + """LoginUseCase 测试""" - @pytest.fixture - def mock_user_repo(self): - """Mock 用户仓储""" - repo = Mock() - repo.find_by_email = Mock(return_value=None) - repo.save = Mock() - repo.get = Mock(return_value=None) - return repo + 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 - @pytest.fixture - def mock_session_store(self): - """Mock Session 存储""" - store = Mock() - store.save_session = Mock() - store.get_refresh_token = Mock(return_value=None) - store.get_session_by_refresh_token = Mock(return_value=None) - store.delete_session = Mock(return_value=True) - store.delete_all_user_sessions = Mock() - return store - - @pytest.fixture - def test_user(self): - """测试用户""" - user = User( - id="user-123", - email="test@example.com", - username="testuser", - display_name="Test User", - password_hash="hashed_password", - ) - return user - - @pytest.fixture - def use_case(self, mock_user_repo, mock_session_store): - """创建登录用例（使用测试用JWT密钥）""" - return LoginUseCase( - user_repository=mock_user_repo, + use_case = LoginUseCase( + mock_user_repo, session_store=mock_session_store, - jwt_secret_key="test-secret-key-for-unit-tests", + jwt_secret_key="test-secret-key-for-jwt-login-123", ) - - def test_login_success(self, use_case, mock_user_repo, mock_session_store, test_user): - """测试登录成功""" - mock_user_repo.find_by_email.return_value = test_user - - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = True - - request = LoginRequest( - email="test@example.com", - password="CorrectPass123", - device_info="Test Device", - ip_address="192.168.1.1", - ) - response, error = use_case.execute(request) + 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-123" + assert response.user_id == "user_001" assert response.email == "test@example.com" assert response.username == "testuser" - assert response.display_name == "Test User" - assert response.access_token != "" - assert response.refresh_token != "" + assert response.display_name == "测试用户" + assert len(response.access_token) > 0 + assert len(response.refresh_token) > 0 assert response.expires_in > 0 - - # 验证 session 已保存 mock_session_store.save_session.assert_called_once() - save_args = mock_session_store.save_session.call_args[1] - assert save_args["user_id"] == "user-123" - assert save_args["device_info"] == "Test Device" - assert save_args["ip_address"] == "192.168.1.1" + mock_user_repo.save.assert_called() # 更新最后登录时间 - # 验证最后登录信息已更新 - mock_user_repo.save.assert_called() - 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 == "192.168.1.1" - - def test_login_email_empty(self, use_case): - """测试邮箱为空""" - request = LoginRequest(email="", password="password123") + 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 error == "Email is required" + assert "Email is required" in error - def test_login_password_empty(self, use_case, mock_user_repo): - """测试密码为空""" - mock_user_repo.find_by_email.return_value = Mock() # 即使有用户也应该在密码检查前失败 + 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 error == "Password is required" + assert "Password is required" in error - def test_login_user_not_found(self, use_case, mock_user_repo): - """测试用户不存在""" + def test_login_user_not_found(self, mock_user_repo, mock_session_store): + """用户不存在返回错误""" mock_user_repo.find_by_email.return_value = None - request = LoginRequest(email="nonexistent@example.com", password="password123") + 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 error == "Invalid email or password" + assert "Invalid email or password" in error - def test_login_wrong_password(self, use_case, mock_user_repo, test_user): - """测试密码错误""" - mock_user_repo.find_by_email.return_value = test_user + def test_login_wrong_password(self, mock_user_repo, mock_session_store, sample_user): + """密码错误返回错误""" + mock_user_repo.find_by_email.return_value = sample_user - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = False - - request = LoginRequest(email="test@example.com", password="WrongPass") - response, error = use_case.execute(request) + 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 error == "Invalid email or password" + assert "Invalid email or password" in error + mock_session_store.save_session.assert_not_called() - def test_login_legacy_sha256_password_success_and_upgrade(self, use_case, mock_user_repo, mock_session_store): - """测试旧版 SHA256 密码登录成功并自动升级哈希""" - legacy_hash = _legacy_sha256("OldPassword123") - legacy_user = User( - id="user-legacy", - email="legacy@example.com", - username="legacyuser", - display_name="Legacy User", - password_hash=legacy_hash, + 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 - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = False # 现代哈希验证失败 - mock_hasher.hash_password.return_value = "new_bcrypt_hash" + saved_user = None - request = LoginRequest(email="legacy@example.com", password="OldPassword123") - response, error = use_case.execute(request) + 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 - assert response.user_id == "user-legacy" + # 密码哈希应该被更新为 bcrypt 格式 + assert saved_user is not None + assert saved_user.password_hash != original_hash + assert saved_user.password_hash.startswith("$2") # bcrypt 格式 - # 验证密码哈希已升级 - mock_user_repo.save.assert_called() - saved_user = mock_user_repo.save.call_args[0][0] - assert saved_user.password_hash == "new_bcrypt_hash" - - def test_login_legacy_sha256_password_wrong(self, use_case, mock_user_repo): - """测试旧版 SHA256 密码错误""" - legacy_hash = _legacy_sha256("CorrectPassword") - legacy_user = User( - id="user-legacy", - email="legacy@example.com", - username="legacyuser", - display_name="Legacy User", - password_hash=legacy_hash, - ) + 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 - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = False - - request = LoginRequest(email="legacy@example.com", password="WrongPassword") - response, error = use_case.execute(request) + 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 error == "Invalid email or password" + assert "Invalid email or password" in error - def test_login_email_normalized_to_lowercase(self, use_case, mock_user_repo, test_user): - """测试邮箱自动转小写并去空格""" - mock_user_repo.find_by_email.return_value = test_user + def test_login_exception_returns_error(self, mock_user_repo, mock_session_store): + """异常时返回友好错误""" + mock_user_repo.find_by_email.side_effect = Exception("DB error") - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = True + 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) - request = LoginRequest(email=" TEST@Example.COM ", password="pass123") - response, error = use_case.execute(request) - - assert error is None - assert response is not None - # find_by_email 应该收到小写去空格后的邮箱 - mock_user_repo.find_by_email.assert_called_with("test@example.com") - - def test_login_default_device_and_ip(self, use_case, mock_user_repo, mock_session_store, test_user): - """测试设备信息和IP的默认值""" - mock_user_repo.find_by_email.return_value = test_user - - with patch("packages.application.auth.login_use_case.password_hasher") as mock_hasher: - mock_hasher.verify_password.return_value = True - - request = LoginRequest(email="test@example.com", password="pass123") - response, error = use_case.execute(request) - - assert error is None - save_args = mock_session_store.save_session.call_args[1] - assert save_args["device_info"] == "Unknown" - assert save_args["ip_address"] == "unknown" + assert response is None + assert "Login failed" in error class TestRefreshTokenUseCase: - """刷新令牌用例测试""" + """RefreshTokenUseCase 测试""" - @pytest.fixture - def mock_user_repo(self): - repo = Mock() - repo.get = Mock(return_value=None) - return repo + 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 - @pytest.fixture - def mock_session_store(self): - store = Mock() - store.get_session_by_refresh_token = Mock(return_value=None) - store.get_refresh_token = Mock(return_value=None) - return store - - @pytest.fixture - def test_user(self): - return User( - id="user-123", - email="test@example.com", - username="testuser", - display_name="Test User", - password_hash="hashed", - ) - - @pytest.fixture - def use_case(self, mock_user_repo, mock_session_store): - # 用 patch 替换 jwt_service.config - with patch("packages.application.auth.login_use_case.jwt_service") as mock_jwt: - mock_jwt.config.SECRET_KEY = "test-secret-key" - mock_jwt.config.ALGORITHM = "HS256" - mock_jwt.config.ACCESS_TOKEN_EXPIRE_MINUTES = 30 - uc = RefreshTokenUseCase( - user_repository=mock_user_repo, - session_store=mock_session_store, - ) - uc._jwt_secret_key = "test-secret-key" - uc.jwt_service.config = mock_jwt.config - yield uc - - def test_refresh_success(self, use_case, mock_user_repo, mock_session_store, test_user): - """测试刷新令牌成功""" - mock_session_store.get_session_by_refresh_token.return_value = { - "session_id": "sess-abc", - "user_id": "user-123", - } - mock_session_store.get_refresh_token.return_value = "valid-refresh-token" - mock_user_repo.get.return_value = test_user - - request = RefreshTokenRequest(refresh_token="valid-refresh-token") + 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-123" - assert response.email == "test@example.com" - assert response.access_token != "" - assert response.refresh_token == "valid-refresh-token" + assert response.user_id == "user_001" + assert len(response.access_token) > 0 + assert response.refresh_token == "valid_refresh_token" # 不变 - def test_refresh_token_empty(self, use_case): - """测试 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 error == "Refresh token is required" + assert "Refresh token is required" in error - def test_refresh_invalid_token(self, use_case, mock_session_store): - """测试无效的 refresh_token""" + 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 - request = RefreshTokenRequest(refresh_token="invalid-token") + 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 error == "Invalid or expired refresh token" + assert "Invalid or expired" in error - def test_refresh_session_data_invalid(self, use_case, mock_session_store): - """测试 session 数据不完整""" - mock_session_store.get_session_by_refresh_token.return_value = { - "session_id": "sess-abc", - # 缺少 user_id - } + 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" - request = RefreshTokenRequest(refresh_token="some-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 error == "Invalid session data" + assert "mismatch" in error - def test_refresh_token_mismatch(self, use_case, mock_user_repo, mock_session_store, test_user): - """测试 refresh_token 不匹配""" - mock_session_store.get_session_by_refresh_token.return_value = { - "session_id": "sess-abc", - "user_id": "user-123", - } - mock_session_store.get_refresh_token.return_value = "different-token" - mock_user_repo.get.return_value = test_user - - request = RefreshTokenRequest(refresh_token="user-provided-token") - response, error = use_case.execute(request) - - assert response is None - assert error == "Refresh token mismatch" - - def test_refresh_user_not_found(self, use_case, mock_user_repo, mock_session_store): - """测试用户不存在""" - mock_session_store.get_session_by_refresh_token.return_value = { - "session_id": "sess-abc", - "user_id": "user-nonexistent", - } - mock_session_store.get_refresh_token.return_value = "valid-token" + 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 - request = RefreshTokenRequest(refresh_token="valid-token") + 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 error == "User not found" + 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" class TestLogoutUseCase: - """登出用例测试""" + """LogoutUseCase 测试""" - @pytest.fixture - def mock_session_store(self): - store = Mock() - store.delete_session = Mock(return_value=True) - store.delete_all_user_sessions = Mock() - return store + def test_logout_single_session(self, mock_session_store): + """单设备登出成功""" + mock_session_store.delete_session.return_value = True - @pytest.fixture - def use_case(self, mock_session_store): - return LogoutUseCase(session_store=mock_session_store) - - def test_logout_single_device_success(self, use_case, mock_session_store): - """测试单设备登出成功""" - request = LogoutRequest(user_id="user-123", session_id="sess-abc") + 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 error is None - mock_session_store.delete_session.assert_called_once_with("sess-abc") - mock_session_store.delete_all_user_sessions.assert_not_called() + mock_session_store.delete_session.assert_called_once_with("sess_123") - def test_logout_all_devices(self, use_case, mock_session_store): - """测试所有设备登出""" - request = LogoutRequest(user_id="user-123", logout_all_devices=True) + 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) assert success is True assert error is None - mock_session_store.delete_all_user_sessions.assert_called_once_with("user-123") - mock_session_store.delete_session.assert_not_called() + mock_session_store.delete_all_user_sessions.assert_called_once_with("user_001") - def test_logout_missing_session_id(self, use_case): - """测试缺少 session_id""" - request = LogoutRequest(user_id="user-123", session_id=None) + 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) assert success is False - assert error == "Session ID is required" + assert "Session ID is required" in error - def test_logout_session_not_found(self, use_case, mock_session_store): - """测试 session 不存在""" + def test_logout_session_not_found(self, mock_session_store): + """session 不存在返回错误""" mock_session_store.delete_session.return_value = False - request = LogoutRequest(user_id="user-123", session_id="nonexistent-sess") + 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 error == "Session not found" + assert "Session not found" in error + + def test_logout_exception_returns_error(self, mock_session_store): + """异常时返回友好错误""" + mock_session_store.delete_session.side_effect = Exception("Redis error") + + 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 False + assert "Logout failed" in error