"""
|
静默模式服务测试
|
"""
|
|
import time
|
import pytest
|
from unittest.mock import Mock, patch
|
|
from app.services.silence_service import SilenceService
|
|
|
class TestSilenceService:
|
"""静默模式服务测试类"""
|
|
def setup_method(self):
|
"""测试前准备"""
|
self.service = SilenceService()
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_activate_silence_mode_success(self, mock_redis_queue, mock_settings):
|
"""测试成功激活群组静默模式"""
|
# 模拟配置
|
mock_settings.silence_mode_enabled = True
|
mock_settings.silence_duration_minutes = 10
|
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
mock_redis_client.setex.return_value = True
|
|
# 测试激活群组静默模式
|
group_id = "test_group@chatroom"
|
result = self.service.activate_silence_mode(group_id)
|
|
assert result is True
|
# 验证Redis调用
|
assert mock_redis_client.setex.call_count == 2 # 设置两个键
|
|
@patch('app.services.silence_service.settings')
|
def test_activate_silence_mode_disabled(self, mock_settings):
|
"""测试静默模式功能禁用时的行为"""
|
mock_settings.silence_mode_enabled = False
|
|
group_id = "test_group@chatroom"
|
result = self.service.activate_silence_mode(group_id)
|
assert result is False
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_is_silence_active(self, mock_redis_queue, mock_settings):
|
"""测试检查静默模式是否激活"""
|
mock_settings.silence_mode_enabled = True
|
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
|
# 测试静默模式激活状态
|
mock_redis_client.exists.return_value = True
|
group_id = "test_group@chatroom"
|
result = self.service.is_silence_active(group_id)
|
assert result is True
|
|
# 测试静默模式未激活状态
|
mock_redis_client.exists.return_value = False
|
result = self.service.is_silence_active(group_id)
|
assert result is False
|
|
@patch('app.services.silence_service.settings')
|
def test_is_silence_active_disabled(self, mock_settings):
|
"""测试功能禁用时的静默模式检查"""
|
mock_settings.silence_mode_enabled = False
|
|
group_id = "test_group@chatroom"
|
result = self.service.is_silence_active(group_id)
|
assert result is False
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_get_silence_remaining_time(self, mock_redis_queue, mock_settings):
|
"""测试获取静默模式剩余时间"""
|
mock_settings.silence_mode_enabled = True
|
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
|
# 模拟静默模式激活
|
mock_redis_client.exists.return_value = True
|
|
# 模拟结束时间（当前时间 + 300秒）
|
future_time = time.time() + 300
|
mock_redis_client.get.return_value = str(future_time)
|
|
group_id = "test_group@chatroom"
|
result = self.service.get_silence_remaining_time(group_id)
|
assert result is not None
|
assert result > 0
|
assert result <= 300
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_get_silence_remaining_time_inactive(self, mock_redis_queue, mock_settings):
|
"""测试静默模式未激活时获取剩余时间"""
|
mock_settings.silence_mode_enabled = True
|
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
mock_redis_client.exists.return_value = False
|
|
group_id = "test_group@chatroom"
|
result = self.service.get_silence_remaining_time(group_id)
|
assert result is None
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_deactivate_silence_mode(self, mock_redis_queue, mock_settings):
|
"""测试手动停用静默模式"""
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
mock_redis_client.delete.return_value = True
|
|
result = self.service.deactivate_silence_mode()
|
assert result is True
|
|
# 验证删除了两个键
|
assert mock_redis_client.delete.call_count == 2
|
|
@patch('app.services.silence_service.settings')
|
@patch('app.services.silence_service.redis_queue')
|
def test_extend_silence_mode(self, mock_redis_queue, mock_settings):
|
"""测试延长静默模式"""
|
mock_settings.silence_mode_enabled = True
|
mock_settings.silence_duration_minutes = 10
|
|
# 模拟Redis操作
|
mock_redis_client = Mock()
|
mock_redis_queue.redis_client = mock_redis_client
|
mock_redis_client.setex.return_value = True
|
|
group_id = "test_group@chatroom"
|
result = self.service.extend_silence_mode(group_id)
|
assert result is True
|
|
@patch('app.services.silence_service.settings')
|
def test_get_silence_status(self, mock_settings):
|
"""测试获取静默模式状态"""
|
mock_settings.silence_mode_enabled = True
|
mock_settings.silence_duration_minutes = 10
|
|
with patch.object(self.service, 'is_silence_active', return_value=False):
|
status = self.service.get_silence_status()
|
|
assert status["enabled"] is True
|
assert status["active"] is False
|
assert status["duration_minutes"] == 10
|
assert status["remaining_seconds"] is None
|
assert status["remaining_minutes"] is None
|
|
@patch('app.services.silence_service.settings')
|
def test_get_silence_status_active(self, mock_settings):
|
"""测试获取激活状态的静默模式状态"""
|
mock_settings.silence_mode_enabled = True
|
mock_settings.silence_duration_minutes = 10
|
|
with patch.object(self.service, 'is_silence_active', return_value=True), \
|
patch.object(self.service, 'get_silence_remaining_time', return_value=300):
|
|
status = self.service.get_silence_status()
|
|
assert status["enabled"] is True
|
assert status["active"] is True
|
assert status["remaining_seconds"] == 300
|
assert status["remaining_minutes"] == 5.0
|
