53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
52 lines
1.3 KiB
Python
52 lines
1.3 KiB
Python
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.core.task_tracker import TaskTracker
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_status_refreshes_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = AsyncMock()
|
|
client.hget.return_value = json.dumps({"task_id": "t1", "status": "pending"})
|
|
client.exists.return_value = 1
|
|
|
|
async def fake_get_client():
|
|
return client
|
|
|
|
monkeypatch.setattr(
|
|
"app.core.task_tracker.redis_service.get_client",
|
|
fake_get_client,
|
|
)
|
|
|
|
tracker = TaskTracker()
|
|
ok = await tracker.update_task_status("user-1", "t1", "running")
|
|
|
|
assert ok is True
|
|
client.hset.assert_awaited_once()
|
|
client.expire.assert_awaited_once_with("task:user:user-1:tasks", tracker.task_ttl)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_task_refreshes_ttl_when_hash_remains(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client = AsyncMock()
|
|
client.exists.return_value = 1
|
|
|
|
async def fake_get_client():
|
|
return client
|
|
|
|
monkeypatch.setattr(
|
|
"app.core.task_tracker.redis_service.get_client",
|
|
fake_get_client,
|
|
)
|
|
|
|
tracker = TaskTracker()
|
|
ok = await tracker.remove_task("user-1", "t1")
|
|
|
|
assert ok is True
|
|
client.hdel.assert_awaited_once()
|
|
client.expire.assert_awaited_once_with("task:user:user-1:tasks", tracker.task_ttl)
|