main:删除多余文档并清理项目目录
变更内容： - 移除冗余文档，包括 Grafana 指南、指标对比、修复总结、OpenAPI 规范等。 - 精简项目文档结构，优化 README 文件内容。 - 提升文档层次清晰度，集中核心指南。
This commit is contained in:
401
tests/test_job_manager.py
Normal file
401
tests/test_job_manager.py
Normal file
@@ -0,0 +1,401 @@
|
||||
"""异步任务管理器测试"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from fastapi import status
|
||||
|
||||
from src.functional_scaffold.core.job_manager import (
|
||||
JobManager,
|
||||
get_job_manager,
|
||||
shutdown_job_manager,
|
||||
)
|
||||
from src.functional_scaffold.api.models import JobStatus
|
||||
|
||||
|
||||
class TestJobManager:
|
||||
"""测试 JobManager 类"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_redis(self):
|
||||
"""模拟 Redis 客户端"""
|
||||
mock = AsyncMock()
|
||||
mock.ping = AsyncMock(return_value=True)
|
||||
mock.hset = AsyncMock()
|
||||
mock.hgetall = AsyncMock(return_value={})
|
||||
mock.expire = AsyncMock()
|
||||
mock.close = AsyncMock()
|
||||
return mock
|
||||
|
||||
@pytest.fixture
|
||||
def mock_http_client(self):
|
||||
"""模拟 HTTP 客户端"""
|
||||
mock = AsyncMock()
|
||||
mock.post = AsyncMock()
|
||||
mock.aclose = AsyncMock()
|
||||
return mock
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_job_id(self):
|
||||
"""测试任务 ID 生成"""
|
||||
manager = JobManager()
|
||||
job_id = manager._generate_job_id()
|
||||
|
||||
assert len(job_id) == 12
|
||||
assert all(c in "0123456789abcdef" for c in job_id)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_timestamp(self):
|
||||
"""测试时间戳生成"""
|
||||
manager = JobManager()
|
||||
timestamp = manager._get_timestamp()
|
||||
|
||||
assert "T" in timestamp
|
||||
assert timestamp.endswith("+00:00") or timestamp.endswith("Z")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_available_algorithms(self):
|
||||
"""测试获取可用算法列表"""
|
||||
manager = JobManager()
|
||||
manager._register_algorithms()
|
||||
|
||||
algorithms = manager.get_available_algorithms()
|
||||
|
||||
assert "PrimeChecker" in algorithms
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_is_available_without_redis(self):
|
||||
"""测试 Redis 不可用时的状态"""
|
||||
manager = JobManager()
|
||||
|
||||
assert manager.is_available() is False
|
||||
|
||||
|
||||
class TestJobManagerWithMocks:
|
||||
"""使用 Mock 测试 JobManager"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_job(self):
|
||||
"""测试创建任务"""
|
||||
manager = JobManager()
|
||||
|
||||
# 模拟 Redis
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hset = AsyncMock()
|
||||
manager._redis_client = mock_redis
|
||||
manager._register_algorithms()
|
||||
|
||||
job_id = await manager.create_job(
|
||||
algorithm="PrimeChecker",
|
||||
params={"number": 17},
|
||||
webhook="https://example.com/callback",
|
||||
request_id="test-request-id",
|
||||
)
|
||||
|
||||
assert len(job_id) == 12
|
||||
mock_redis.hset.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_job_invalid_algorithm(self):
|
||||
"""测试创建任务时算法不存在"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
manager._redis_client = mock_redis
|
||||
manager._register_algorithms()
|
||||
|
||||
with pytest.raises(ValueError, match="不存在"):
|
||||
await manager.create_job(
|
||||
algorithm="NonExistentAlgorithm",
|
||||
params={},
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_job_redis_unavailable(self):
|
||||
"""测试 Redis 不可用时创建任务"""
|
||||
manager = JobManager()
|
||||
manager._register_algorithms()
|
||||
|
||||
with pytest.raises(RuntimeError, match="Redis 不可用"):
|
||||
await manager.create_job(
|
||||
algorithm="PrimeChecker",
|
||||
params={"number": 17},
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_job(self):
|
||||
"""测试获取任务信息"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hgetall = AsyncMock(
|
||||
return_value={
|
||||
"status": "completed",
|
||||
"algorithm": "PrimeChecker",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
"started_at": "2026-02-02T10:00:01+00:00",
|
||||
"completed_at": "2026-02-02T10:00:02+00:00",
|
||||
"result": '{"number": 17, "is_prime": true}',
|
||||
"error": "",
|
||||
"metadata": '{"elapsed_time": 0.001}',
|
||||
}
|
||||
)
|
||||
manager._redis_client = mock_redis
|
||||
|
||||
job_data = await manager.get_job("test-job-id")
|
||||
|
||||
assert job_data is not None
|
||||
assert job_data["job_id"] == "test-job-id"
|
||||
assert job_data["status"] == "completed"
|
||||
assert job_data["algorithm"] == "PrimeChecker"
|
||||
assert job_data["result"]["number"] == 17
|
||||
assert job_data["result"]["is_prime"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_job_not_found(self):
|
||||
"""测试获取不存在的任务"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hgetall = AsyncMock(return_value={})
|
||||
manager._redis_client = mock_redis
|
||||
|
||||
job_data = await manager.get_job("non-existent-job")
|
||||
|
||||
assert job_data is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_job(self):
|
||||
"""测试执行任务"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hgetall = AsyncMock(
|
||||
return_value={
|
||||
"status": "pending",
|
||||
"algorithm": "PrimeChecker",
|
||||
"params": '{"number": 17}',
|
||||
"webhook": "",
|
||||
"request_id": "test-request-id",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
}
|
||||
)
|
||||
mock_redis.hset = AsyncMock()
|
||||
mock_redis.expire = AsyncMock()
|
||||
manager._redis_client = mock_redis
|
||||
manager._register_algorithms()
|
||||
|
||||
await manager.execute_job("test-job-id")
|
||||
|
||||
# 验证状态更新被调用
|
||||
assert mock_redis.hset.call_count >= 2 # running + completed
|
||||
mock_redis.expire.assert_called_once()
|
||||
|
||||
|
||||
class TestJobsAPI:
|
||||
"""测试 /jobs API 端点"""
|
||||
|
||||
def test_create_job_success(self, client):
|
||||
"""测试成功创建任务"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = True
|
||||
mock_manager.get_available_algorithms.return_value = ["PrimeChecker"]
|
||||
mock_manager.create_job = AsyncMock(return_value="abc123def456")
|
||||
mock_manager.get_job = AsyncMock(
|
||||
return_value={
|
||||
"job_id": "abc123def456",
|
||||
"status": "pending",
|
||||
"algorithm": "PrimeChecker",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
}
|
||||
)
|
||||
mock_manager.execute_job = AsyncMock()
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.post(
|
||||
"/jobs",
|
||||
json={
|
||||
"algorithm": "PrimeChecker",
|
||||
"params": {"number": 17},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_202_ACCEPTED
|
||||
data = response.json()
|
||||
assert data["job_id"] == "abc123def456"
|
||||
assert data["status"] == "pending"
|
||||
assert data["message"] == "任务已创建"
|
||||
|
||||
def test_create_job_algorithm_not_found(self, client):
|
||||
"""测试创建任务时算法不存在"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = True
|
||||
mock_manager.get_available_algorithms.return_value = ["PrimeChecker"]
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.post(
|
||||
"/jobs",
|
||||
json={
|
||||
"algorithm": "NonExistentAlgorithm",
|
||||
"params": {},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "ALGORITHM_NOT_FOUND"
|
||||
|
||||
def test_create_job_service_unavailable(self, client):
|
||||
"""测试服务不可用时创建任务"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = False
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.post(
|
||||
"/jobs",
|
||||
json={
|
||||
"algorithm": "PrimeChecker",
|
||||
"params": {"number": 17},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
|
||||
|
||||
def test_get_job_status_success(self, client):
|
||||
"""测试成功查询任务状态"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = True
|
||||
mock_manager.get_job = AsyncMock(
|
||||
return_value={
|
||||
"job_id": "abc123def456",
|
||||
"status": "completed",
|
||||
"algorithm": "PrimeChecker",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
"started_at": "2026-02-02T10:00:01+00:00",
|
||||
"completed_at": "2026-02-02T10:00:02+00:00",
|
||||
"result": {"number": 17, "is_prime": True},
|
||||
"error": None,
|
||||
"metadata": {"elapsed_time": 0.001},
|
||||
}
|
||||
)
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.get("/jobs/abc123def456")
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["job_id"] == "abc123def456"
|
||||
assert data["status"] == "completed"
|
||||
assert data["result"]["is_prime"] is True
|
||||
|
||||
def test_get_job_status_not_found(self, client):
|
||||
"""测试查询不存在的任务"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = True
|
||||
mock_manager.get_job = AsyncMock(return_value=None)
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.get("/jobs/non-existent-job")
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "JOB_NOT_FOUND"
|
||||
|
||||
def test_get_job_status_service_unavailable(self, client):
|
||||
"""测试服务不可用时查询任务"""
|
||||
with patch(
|
||||
"src.functional_scaffold.api.routes.get_job_manager", new_callable=AsyncMock
|
||||
) as mock_get_manager:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.is_available.return_value = False
|
||||
mock_get_manager.return_value = mock_manager
|
||||
|
||||
response = client.get("/jobs/abc123def456")
|
||||
|
||||
assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
|
||||
|
||||
|
||||
class TestWebhook:
|
||||
"""测试 Webhook 回调"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_webhook_success(self):
|
||||
"""测试成功发送 Webhook"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hgetall = AsyncMock(
|
||||
return_value={
|
||||
"status": "completed",
|
||||
"algorithm": "PrimeChecker",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
"completed_at": "2026-02-02T10:00:02+00:00",
|
||||
"result": '{"number": 17, "is_prime": true}',
|
||||
"error": "",
|
||||
"metadata": '{"elapsed_time": 0.001}',
|
||||
}
|
||||
)
|
||||
manager._redis_client = mock_redis
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
|
||||
mock_http = AsyncMock()
|
||||
mock_http.post = AsyncMock(return_value=mock_response)
|
||||
manager._http_client = mock_http
|
||||
|
||||
await manager._send_webhook("test-job-id", "https://example.com/callback")
|
||||
|
||||
mock_http.post.assert_called_once()
|
||||
call_args = mock_http.post.call_args
|
||||
assert call_args[0][0] == "https://example.com/callback"
|
||||
assert "json" in call_args[1]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_webhook_retry_on_failure(self):
|
||||
"""测试 Webhook 失败时重试"""
|
||||
manager = JobManager()
|
||||
|
||||
mock_redis = AsyncMock()
|
||||
mock_redis.hgetall = AsyncMock(
|
||||
return_value={
|
||||
"status": "completed",
|
||||
"algorithm": "PrimeChecker",
|
||||
"created_at": "2026-02-02T10:00:00+00:00",
|
||||
"completed_at": "2026-02-02T10:00:02+00:00",
|
||||
"result": "{}",
|
||||
"error": "",
|
||||
"metadata": "{}",
|
||||
}
|
||||
)
|
||||
manager._redis_client = mock_redis
|
||||
|
||||
mock_http = AsyncMock()
|
||||
mock_http.post = AsyncMock(side_effect=Exception("Connection error"))
|
||||
manager._http_client = mock_http
|
||||
|
||||
# 使用较短的重试间隔进行测试
|
||||
with patch("src.functional_scaffold.core.job_manager.settings") as mock_settings:
|
||||
mock_settings.webhook_max_retries = 2
|
||||
mock_settings.webhook_timeout = 1
|
||||
|
||||
await manager._send_webhook("test-job-id", "https://example.com/callback")
|
||||
|
||||
# 验证重试次数
|
||||
assert mock_http.post.call_count == 2
|
||||
Reference in New Issue
Block a user