""" Redis Session 存储测试 """ import json from unittest.mock import Mock import pytest from packages.adapters.redis.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"