xiaoxia-saas/tests/unit/test_session_store.py

"""
Redis Session 存储测试
"""
import pytest
import json
from datetime import datetime
from unittest.mock import Mock, MagicMock

from packages.domain.auth.session_store import SessionStore


class TestSessionStore:
    """Session 存储测试"""
    
    @pytest.fixture
    def mock_redis(self):
        """创建 Mock Redis 客户端"""
        redis_mock = Mock()
        redis_mock.data = {}  # 模拟内存存储
        redis_mock.expires = {}  # 模拟过期时间
        redis_mock.sets = {}  # 模拟集合
        
        def setex(key, seconds, value):
            redis_mock.data[key] = value
            redis_mock.expires[key] = seconds
            return True
        
        def get(key):
            return redis_mock.data.get(key)
        
        def delete(key):
            if key in redis_mock.data:
                del redis_mock.data[key]
                return 1
            return 0
        
        def exists(key):
            return 1 if key in redis_mock.data else 0
        
        def ttl(key):
            return redis_mock.expires.get(key, -1)
        
        def sadd(key, *values):
            if key not in redis_mock.sets:
                redis_mock.sets[key] = set()
            redis_mock.sets[key].update(values)
            return len(values)
        
        def smembers(key):
            return redis_mock.sets.get(key, set())
        
        def srem(key, *values):
            if key in redis_mock.sets:
                redis_mock.sets[key].discard(*values)
                return len(values)
            return 0
        
        def expire(key, seconds):
            redis_mock.expires[key] = seconds
            return True
        
        redis_mock.setex = setex
        redis_mock.get = get
        redis_mock.delete = delete
        redis_mock.exists = exists
        redis_mock.ttl = ttl
        redis_mock.sadd = sadd
        redis_mock.smembers = smembers
        redis_mock.srem = srem
        redis_mock.expire = expire
        
        return redis_mock
    
    @pytest.fixture
    def session_store(self, mock_redis):
        """创建 Session 存储实例"""
        return SessionStore(redis_client=mock_redis)
    
    def test_save_session(self, session_store, mock_redis):
        """测试保存 Session"""
        result = session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="refresh-token-abc",
            device_info="Chrome/Windows",
            ip_address="192.168.1.1",
            expires_in_seconds=3600,
        )
        
        assert result is True
        
        # 验证数据已保存
        session_key = "session:session-123"
        assert session_key in mock_redis.data
        
        session_data = json.loads(mock_redis.data[session_key])
        assert session_data["session_id"] == "session-123"
        assert session_data["user_id"] == "user-456"
        assert session_data["device_info"] == "Chrome/Windows"
        assert session_data["ip_address"] == "192.168.1.1"
        
        # 验证 refresh_token 已保存
        refresh_token_key = "refresh_token:session-123"
        assert mock_redis.data[refresh_token_key] == "refresh-token-abc"
        
        # 验证用户 Session 集合已更新
        user_sessions_key = "user_sessions:user-456"
        assert "session-123" in mock_redis.sets[user_sessions_key]
    
    def test_get_session(self, session_store, mock_redis):
        """测试获取 Session"""
        # 先保存
        session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="token",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        # 获取
        session = session_store.get_session("session-123")
        
        assert session is not None
        assert session["session_id"] == "session-123"
        assert session["user_id"] == "user-456"
        assert session["device_info"] == "Chrome"
    
    def test_get_nonexistent_session(self, session_store):
        """测试获取不存在的 Session"""
        session = session_store.get_session("nonexistent")
        assert session is None
    
    def test_get_refresh_token(self, session_store):
        """测试获取 refresh_token"""
        session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="my-refresh-token",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        token = session_store.get_refresh_token("session-123")
        assert token == "my-refresh-token"
    
    def test_update_last_active(self, session_store, mock_redis):
        """测试更新最后活跃时间"""
        session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="token",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        # 获取原始时间
        session1 = session_store.get_session("session-123")
        original_time = session1["last_active_at"]
        
        # 更新
        result = session_store.update_last_active("session-123")
        assert result is True
        
        # 验证时间已更新
        session2 = session_store.get_session("session-123")
        assert session2["last_active_at"] >= original_time
    
    def test_delete_session(self, session_store, mock_redis):
        """测试删除 Session"""
        session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="token",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        # 删除
        result = session_store.delete_session("session-123")
        assert result is True
        
        # 验证已删除
        session = session_store.get_session("session-123")
        assert session is None
        
        token = session_store.get_refresh_token("session-123")
        assert token is None
        
        # 验证从用户集合中移除
        user_sessions_key = "user_sessions:user-456"
        assert "session-123" not in mock_redis.sets.get(user_sessions_key, set())
    
    def test_get_user_sessions(self, session_store):
        """测试获取用户的所有 Session"""
        # 创建多个 Session
        session_store.save_session(
            session_id="session-1",
            user_id="user-456",
            refresh_token="token1",
            device_info="Chrome",
            ip_address="192.168.1.1",
        )
        
        session_store.save_session(
            session_id="session-2",
            user_id="user-456",
            refresh_token="token2",
            device_info="Firefox",
            ip_address="192.168.1.2",
        )
        
        # 获取
        sessions = session_store.get_user_sessions("user-456")
        
        assert len(sessions) == 2
        session_ids = [s["session_id"] for s in sessions]
        assert "session-1" in session_ids
        assert "session-2" in session_ids
    
    def test_delete_all_user_sessions(self, session_store):
        """测试删除用户的所有 Session"""
        # 创建多个 Session
        session_store.save_session(
            session_id="session-1",
            user_id="user-456",
            refresh_token="token1",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        session_store.save_session(
            session_id="session-2",
            user_id="user-456",
            refresh_token="token2",
            device_info="Firefox",
            ip_address="127.0.0.1",
        )
        
        # 删除所有
        count = session_store.delete_all_user_sessions("user-456")
        assert count == 2
        
        # 验证已删除
        sessions = session_store.get_user_sessions("user-456")
        assert len(sessions) == 0
    
    def test_session_exists(self, session_store):
        """测试检查 Session 是否存在"""
        assert session_store.session_exists("nonexistent") is False
        
        session_store.save_session(
            session_id="session-123",
            user_id="user-456",
            refresh_token="token",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        assert session_store.session_exists("session-123") is True
    
    def test_multiple_users(self, session_store):
        """测试多用户隔离"""
        # 用户 1 的 Session
        session_store.save_session(
            session_id="session-user1",
            user_id="user-1",
            refresh_token="token1",
            device_info="Chrome",
            ip_address="127.0.0.1",
        )
        
        # 用户 2 的 Session
        session_store.save_session(
            session_id="session-user2",
            user_id="user-2",
            refresh_token="token2",
            device_info="Firefox",
            ip_address="127.0.0.1",
        )
        
        # 验证隔离
        user1_sessions = session_store.get_user_sessions("user-1")
        assert len(user1_sessions) == 1
        assert user1_sessions[0]["session_id"] == "session-user1"
        
        user2_sessions = session_store.get_user_sessions("user-2")
        assert len(user2_sessions) == 1
        assert user2_sessions[0]["session_id"] == "session-user2"