test_async_redis_service.py 1011 B

123456789101112131415161718192021222324252627282930313233
  1. import pytest
  2. import asyncio
  3. from config import settings
  4. from services.async_redis_service import AsyncRedisService
  5. from core.base.async_redis_client import RedisManager
  6. @pytest.mark.asyncio
  7. async def test_async_redis_service_mark_and_get_status():
  8. await RedisManager.init(redis_url=settings.redis_url)
  9. service = AsyncRedisService(prefix="crawler:task", ttl=20)
  10. test_message_id = "test_123456"
  11. # 确保干净
  12. pool = RedisManager.get_pool()
  13. await pool.delete(service._build_key(test_message_id))
  14. # 初始应获取 None
  15. status = await service.get_status(test_message_id)
  16. assert status is None
  17. # 标记为执行中
  18. await service.mark_processing(test_message_id)
  19. status = await service.get_status(test_message_id)
  20. assert status == "0"
  21. # 标记为完成
  22. await service.mark_done(test_message_id)
  23. status = await service.get_status(test_message_id)
  24. assert status == "1"
  25. # 清理
  26. # await pool.delete(service._build_key(test_message_id))