xiaoxia-saas/tests/unit/test_login_use_case.py

"""登录 Use Case 单元测试."""

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 (
    LEGACY_SHA256_HEX_LENGTH,
    LoginRequest,
    LoginResponse,
    LoginUseCase,
    LogoutRequest,
    LogoutUseCase,
    RefreshTokenRequest,
    RefreshTokenUseCase,
    _is_legacy_sha256_hash,
    _legacy_sha256,
)
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():
    repo = MagicMock()
    repo.find_by_email.return_value = None
    repo.save.return_value = None
    return repo


@pytest.fixture
def mock_session_store():
    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


# ── _is_legacy_sha256_hash 测试 ────────────────────────────────────────────


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


# ── _legacy_sha256 测试 ────────────────────────────────────────────────────


class TestLegacySha256:
    """legacy SHA-256 哈希函数测试"""

    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_deterministic(self):
        """相同输入产生相同输出."""
        assert _legacy_sha256("test") == _legacy_sha256("test")

    def test_different_inputs_different_outputs(self):
        """不同输入产生不同输出."""
        assert _legacy_sha256("pass1") != _legacy_sha256("pass2")

    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:
    """登录请求数据类测试"""

    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="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="u@e.com", password="pass")
        assert req.ip_address == "unknown"

    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"


# ── LoginResponse 测试 ─────────────────────────────────────────────────────


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,
        )
        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


# ── LogoutUseCase 测试 ─────────────────────────────────────────────────────


class TestLogoutUseCase:
    """登出用例测试"""

    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)

        assert ok is True
        assert error is None
        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):
        """登出所有设备."""
        uc = LogoutUseCase(session_store=mock_session_store)
        req = LogoutRequest(user_id="user-1", logout_all_devices=True)
        ok, error = uc.execute(req)

        assert ok is True
        assert error is None
        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_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 ok is False
        assert "Session ID is required" in error

    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)

        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 down")
        uc = LogoutUseCase(session_store=mock_session_store)
        req = LogoutRequest(user_id="user-1", session_id="sess-1")
        ok, error = uc.execute(req)

        assert ok is False
        assert "failed" in error.lower()

    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