Simon
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
# =============================================================================
|
|
# Token异常检测 - 单元测试
|
|
# =============================================================================
|
|
# 测试用例：
|
|
# T001: record_token_ip 正常记录
|
|
# T002: get_token_ips 获取IP列表
|
|
# T003: 多IP记录超过阈值触发告警
|
|
# T004: IP记录自动过期
|
|
# T005: 异常处理 - Redis连接失败
|
|
# T006: 异常处理 - 无效token
|
|
# =============================================================================
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime
|
|
|
|
|
|
class TestTokenServiceIPRecord:
|
|
"""Token IP记录功能测试"""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def mock_redis(self):
|
|
"""创建模拟Redis客户端"""
|
|
mock = AsyncMock()
|
|
# 模拟setex和get方法
|
|
mock.setex = AsyncMock(return_value=True)
|
|
mock.get = AsyncMock(return_value=None)
|
|
return mock
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_token_ip(self, mock_redis):
|
|
"""T001: record_token_ip 正常记录"""
|
|
from app.services.token_service import TokenService
|
|
|
|
service = TokenService(mock_redis)
|
|
|
|
# 测试记录Token IP
|
|
await service.record_token_ip("test_token_123", "192.168.1.100")
|
|
|
|
# 验证Redis setex被调用
|
|
mock_redis.setex.assert_called_once()
|
|
call_args = mock_redis.setex.call_args
|
|
assert call_args[0][0].startswith("token_ip:")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_token_ips(self, mock_redis):
|
|
"""T002: get_token_ips 获取IP列表"""
|
|
from app.services.token_service import TokenService
|
|
|
|
# 模拟已有IP记录
|
|
mock_redis.get = AsyncMock(
|
|
return_value="192.168.1.100@2026-07-14T10:00:00,192.168.1.101@2026-07-14T10:05:00"
|
|
)
|
|
|
|
service = TokenService(mock_redis)
|
|
result = await service.get_token_ips("test_token_123")
|
|
|
|
assert len(result) == 2
|
|
assert result[0]["ip"] == "192.168.1.100"
|
|
assert result[1]["ip"] == "192.168.1.101"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_token_ips_empty(self, mock_redis):
|
|
"""T002-2: get_token_ips 无记录返回空列表"""
|
|
from app.services.token_service import TokenService
|
|
|
|
mock_redis.get = AsyncMock(return_value=None)
|
|
|
|
service = TokenService(mock_redis)
|
|
result = await service.get_token_ips("nonexistent_token")
|
|
|
|
assert result == []
|
|
|
|
|
|
class TestTokenAnomalyDetection:
|
|
"""Token异常检测测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anomaly_detection_threshold(self):
|
|
"""T003: 多IP记录超过阈值"""
|
|
# 模拟3个不同IP的记录
|
|
test_ips = ["192.168.1.100", "192.168.1.101", "192.168.1.102"]
|
|
|
|
# 验证阈值判断
|
|
ip_count = len(set(test_ips))
|
|
assert ip_count >= 3 # 触发阈值
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_usage_no_anomaly(self):
|
|
"""T003-2: 正常使用不触发告警"""
|
|
# 单个IP正常使用
|
|
test_ips = ["192.168.1.100"]
|
|
|
|
ip_count = len(set(test_ips))
|
|
assert ip_count < 3 # 不触发
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_ips_no_anomaly(self):
|
|
"""T003-3: 两个IP不触发告警"""
|
|
test_ips = ["192.168.1.100", "192.168.1.101"]
|
|
|
|
ip_count = len(set(test_ips))
|
|
assert ip_count < 3 # 不触发
|
|
|
|
|
|
class TestTokenAnomalyDetectionTask:
|
|
"""Token异常检测定时任务测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_anomaly_logic(self):
|
|
"""T004: 异常检测逻辑验证"""
|
|
from app.tasks.token_anomaly_detection import TOKEN_ANOMALY_IP_THRESHOLD
|
|
|
|
# 测试阈值配置
|
|
assert TOKEN_ANOMALY_IP_THRESHOLD == 3
|
|
|
|
# 验证异常检测逻辑
|
|
test_cases = [
|
|
(["10.0.0.1", "10.0.0.2", "10.0.0.3"], True), # 3个IP，触发
|
|
(["10.0.0.1", "10.0.0.2"], False), # 2个IP，不触发
|
|
(["10.0.0.1"], False), # 1个IP，不触发
|
|
]
|
|
|
|
for ips, expected in test_cases:
|
|
result = len(set(ips)) >= TOKEN_ANOMALY_IP_THRESHOLD
|
|
assert result == expected
|
|
|
|
|
|
# 运行测试
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|