""" 好友忽略服务测试 """ import pytest from unittest.mock import Mock, patch from app.services.friend_ignore_service import FriendIgnoreService class TestFriendIgnoreService: """好友忽略服务测试类""" def setup_method(self): """测试前准备""" self.service = FriendIgnoreService() @patch('app.services.friend_ignore_service.redis_queue') def test_add_friends_to_ignore_list_success(self, mock_redis_queue): """测试成功添加好友到忽略列表""" # 模拟Redis操作 mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client friends = ["wxid_test1", "wxid_test2", "wxid_test3"] result = self.service.add_friends_to_ignore_list(friends) # 验证结果 assert result is True # 验证Redis操作被调用 mock_redis_client.delete.assert_called_once_with(self.service.ignore_list_key) mock_redis_client.sadd.assert_called_once_with(self.service.ignore_list_key, *friends) @patch('app.services.friend_ignore_service.redis_queue') def test_add_friends_to_ignore_list_empty(self, mock_redis_queue): """测试添加空好友列表""" result = self.service.add_friends_to_ignore_list([]) # 验证结果 assert result is True # 验证Redis操作未被调用 mock_redis_queue.redis_client.delete.assert_not_called() mock_redis_queue.redis_client.sadd.assert_not_called() @patch('app.services.friend_ignore_service.redis_queue') def test_is_friend_ignored_true(self, mock_redis_queue): """测试检查好友在忽略列表中""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.return_value = True result = self.service.is_friend_ignored("wxid_test1") assert result is True mock_redis_client.sismember.assert_called_once_with(self.service.ignore_list_key, "wxid_test1") @patch('app.services.friend_ignore_service.redis_queue') def test_is_friend_ignored_false(self, mock_redis_queue): """测试检查好友不在忽略列表中""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.return_value = False result = self.service.is_friend_ignored("wxid_test1") assert result is False mock_redis_client.sismember.assert_called_once_with(self.service.ignore_list_key, "wxid_test1") @patch('app.services.friend_ignore_service.redis_queue') def test_get_ignore_list(self, mock_redis_queue): """测试获取忽略列表""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client expected_set = {"wxid_test1", "wxid_test2"} mock_redis_client.smembers.return_value = expected_set result = self.service.get_ignore_list() assert result == expected_set mock_redis_client.smembers.assert_called_once_with(self.service.ignore_list_key) @patch('app.services.friend_ignore_service.redis_queue') def test_remove_friend_from_ignore_list(self, mock_redis_queue): """测试从忽略列表移除好友""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.srem.return_value = 1 # 表示成功移除 result = self.service.remove_friend_from_ignore_list("wxid_test1") assert result is True mock_redis_client.srem.assert_called_once_with(self.service.ignore_list_key, "wxid_test1") @patch('app.services.friend_ignore_service.redis_queue') def test_clear_ignore_list(self, mock_redis_queue): """测试清空忽略列表""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client result = self.service.clear_ignore_list() assert result is True mock_redis_client.delete.assert_called_once_with(self.service.ignore_list_key) @patch('app.services.friend_ignore_service.redis_queue') def test_get_ignore_list_count(self, mock_redis_queue): """测试获取忽略列表数量""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.scard.return_value = 5 result = self.service.get_ignore_list_count() assert result == 5 mock_redis_client.scard.assert_called_once_with(self.service.ignore_list_key) @patch('app.services.friend_ignore_service.redis_queue') def test_add_friends_exception_handling(self, mock_redis_queue): """测试添加好友时的异常处理""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sadd.side_effect = Exception("Redis error") result = self.service.add_friends_to_ignore_list(["wxid_test1"]) assert result is False @patch('app.services.friend_ignore_service.redis_queue') def test_is_friend_ignored_exception_handling(self, mock_redis_queue): """测试检查好友时的异常处理""" mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.side_effect = Exception("Redis error") result = self.service.is_friend_ignored("wxid_test1") assert result is False @patch('app.services.friend_ignore_service.settings') @patch('app.services.friend_ignore_service.redis_queue') @patch('app.services.friend_ignore_service.get_db') def test_is_friend_ignored_whitelist(self, mock_get_db, mock_redis_queue, mock_settings): """测试白名单功能""" # 模拟配置 mock_settings.friend_ignore_enabled = True mock_settings.friend_ignore_whitelist = ["测试用户1", "测试用户2"] # 模拟数据库查询 mock_db = Mock() mock_get_db.return_value.__next__.return_value.__enter__.return_value = mock_db mock_get_db.return_value.__next__.return_value.__exit__.return_value = None # 模拟联系人查询结果 - 根据昵称返回不同的联系人 def mock_query_side_effect(*args, **kwargs): mock_query = Mock() mock_filter = Mock() mock_query.filter.return_value = mock_filter # 根据查询条件返回不同的结果 def mock_first(): # 这里简化处理，假设查询"测试用户1"时返回wxid_whitelist1 mock_contact = Mock() mock_contact.wc_id = "wxid_whitelist1" return mock_contact mock_filter.first = mock_first return mock_query mock_db.query.side_effect = mock_query_side_effect # 模拟Redis操作 mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.return_value = True # 在忽略列表中 # 测试白名单用户（即使在忽略列表中也不应该被忽略） result = self.service.is_friend_ignored("wxid_whitelist1") assert result is False # 测试非白名单用户（在忽略列表中应该被忽略） result = self.service.is_friend_ignored("wxid_normal_user") assert result is True @patch('app.services.friend_ignore_service.settings') def test_is_friend_ignored_disabled(self, mock_settings): """测试功能禁用时的行为""" mock_settings.friend_ignore_enabled = False mock_settings.friend_ignore_whitelist = [] result = self.service.is_friend_ignored("wxid_test1") assert result is False @patch('app.services.friend_ignore_service.settings') @patch('app.services.friend_ignore_service.redis_queue') @patch('app.services.friend_ignore_service.get_db') def test_get_ignore_status_info(self, mock_get_db, mock_redis_queue, mock_settings): """测试获取详细状态信息""" mock_settings.friend_ignore_enabled = True mock_settings.friend_ignore_whitelist = ["测试用户1"] # 模拟数据库查询 mock_db = Mock() mock_get_db.return_value.__enter__.return_value = mock_db mock_get_db.return_value.__exit__.return_value = None # 模拟联系人查询结果 mock_contact = Mock() mock_contact.wc_id = "wxid_whitelist1" mock_db.query.return_value.filter.return_value.first.return_value = mock_contact mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.return_value = True # 测试白名单用户 result = self.service.get_ignore_status_info("wxid_whitelist1") assert result["w_id"] == "wxid_whitelist1" assert result["in_whitelist"] is True assert result["final_ignored"] is False assert "白名单" in result["reason"] assert result["whitelist_nicknames"] == ["测试用户1"] # 测试普通用户 result = self.service.get_ignore_status_info("wxid_normal_user") assert result["w_id"] == "wxid_normal_user" assert result["in_whitelist"] is False assert result["in_ignore_list"] is True assert result["final_ignored"] is True @patch('app.services.friend_ignore_service.FriendIgnoreService._get_whitelist_wids') @patch('app.services.friend_ignore_service.settings') @patch('app.services.friend_ignore_service.redis_queue') def test_is_friend_ignored_nickname_whitelist(self, mock_redis_queue, mock_settings, mock_get_whitelist_wids): """测试昵称白名单功能""" # 模拟配置 mock_settings.friend_ignore_enabled = True mock_settings.friend_ignore_whitelist = ["测试用户1", "测试用户2"] # 模拟白名单w_id转换结果 mock_get_whitelist_wids.return_value = ["wxid_whitelist1", "wxid_whitelist2"] # 模拟Redis操作 mock_redis_client = Mock() mock_redis_queue.redis_client = mock_redis_client mock_redis_client.sismember.return_value = True # 在忽略列表中 # 测试白名单用户（即使在忽略列表中也不应该被忽略） result = self.service.is_friend_ignored("wxid_whitelist1") assert result is False # 测试非白名单用户（在忽略列表中应该被忽略） result = self.service.is_friend_ignored("wxid_normal_user") assert result is True # 验证方法被调用 assert mock_get_whitelist_wids.call_count >= 2