zwf
45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
"""
|
|
Redis 集成测试。
|
|
需要真实 Redis 服务，默认被跳过。
|
|
运行方式：pytest tests/integration/ -v
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from paste.db.redis import Redis
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestRedisConnection:
|
|
"""Redis 集成测试"""
|
|
|
|
@pytest.mark.skip(reason="需要真实 Redis 服务")
|
|
@pytest.mark.asyncio
|
|
async def test_redis_ping(self):
|
|
"""测试 Redis 连通性"""
|
|
result = await Redis.ping()
|
|
assert result is True
|
|
|
|
@pytest.mark.skip(reason="需要真实 Redis 服务")
|
|
@pytest.mark.asyncio
|
|
async def test_redis_get_set(self):
|
|
"""测试 Redis 基本读写"""
|
|
from paste.db.redis import Redis
|
|
|
|
async with await Redis.get_redis() as r:
|
|
# 写入测试
|
|
await r.set("test_key", "test_value")
|
|
# 读取验证
|
|
value = await r.get("test_key")
|
|
assert value == b"test_value"
|
|
# 清理
|
|
await r.delete("test_key")
|
|
|
|
@pytest.mark.skip(reason="需要真实 Redis 服务")
|
|
@pytest.mark.asyncio
|
|
async def test_redis_get_keys(self):
|
|
"""测试获取所有 keys"""
|
|
from paste.db.redis import Redis
|
|
|
|
keys = await Redis.keys()
|
|
assert isinstance(keys, list) |