"""
|
好友忽略服务测试
|
"""
|
|
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
|
