246 lines
9.6 KiB
Python
246 lines
9.6 KiB
Python
"""U3 — SecretsStore Redis 后端单元测试。
|
||
|
||
覆盖场景：
|
||
- Redis 后端 CRUD（set/get/delete/list_keys）
|
||
- Redis=None 降级到内存字典
|
||
- 加密-存储-读取-解密往返（Redis 后端）
|
||
- Redis 连接异常 fail-closed（get_secret 返回 None）
|
||
- 多实例共享语义（两个 store 共用同一 Redis mock）
|
||
|
||
不依赖真实 Redis 实例 — 使用 AsyncMock 模拟 redis.asyncio.Redis。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
|
||
from agentkit.channels.secrets import KEY_SIZE, SecretsStore
|
||
|
||
|
||
# ── 辅助函数 ──────────────────────────────────────────────
|
||
|
||
|
||
def _make_master_key() -> bytes:
|
||
"""确定性测试 master key（32 字节）。"""
|
||
return b"\x01" * KEY_SIZE
|
||
|
||
|
||
class _FakeRedis:
|
||
"""极简 Redis mock — 仅模拟 SecretsStore 用到的方法。
|
||
|
||
内部用 dict 存储，支持 set/get/delete/scan_iter。
|
||
scan_iter 返回 async generator（匹配真实 Redis 接口）。
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._data: dict[str, str] = {}
|
||
|
||
async def set(self, key: str, value: str, **kwargs) -> bool:
|
||
self._data[key] = value
|
||
return True
|
||
|
||
async def get(self, key: str) -> str | None:
|
||
return self._data.get(key)
|
||
|
||
async def delete(self, key: str) -> int:
|
||
if key in self._data:
|
||
del self._data[key]
|
||
return 1
|
||
return 0
|
||
|
||
async def scan_iter(self, match: str = "*"):
|
||
import fnmatch
|
||
|
||
for k in self._data:
|
||
if fnmatch.fnmatch(k, match):
|
||
yield k
|
||
|
||
|
||
def _make_failing_redis() -> AsyncMock:
|
||
"""构造一个所有方法都抛 ConnectionError 的 Redis mock。"""
|
||
redis = AsyncMock()
|
||
redis.get = AsyncMock(side_effect=ConnectionError("redis down"))
|
||
redis.set = AsyncMock(side_effect=ConnectionError("redis down"))
|
||
redis.delete = AsyncMock(side_effect=ConnectionError("redis down"))
|
||
redis.scan_iter = MagicMock(side_effect=ConnectionError("redis down"))
|
||
return redis
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clean_env(monkeypatch):
|
||
"""确保测试不在生产模式下运行。"""
|
||
monkeypatch.delenv("AGENTKIT_ENV", raising=False)
|
||
monkeypatch.delenv("AGENTKIT_MASTER_KEY", raising=False)
|
||
|
||
|
||
# ── Redis 后端 CRUD ──────────────────────────────────────
|
||
|
||
|
||
class TestRedisBackendCrud:
|
||
"""Redis 后端的 set/get/delete/list_keys 操作。"""
|
||
|
||
async def test_set_and_get_secret_with_redis(self):
|
||
"""Redis 后端：写入后读取返回明文。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
await store.set_secret("feishu:app_id", "cli_xxx")
|
||
assert await store.get_secret("feishu:app_id") == "cli_xxx"
|
||
|
||
async def test_get_nonexistent_returns_none_with_redis(self):
|
||
"""Redis 后端：读取不存在的 key 返回 None。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
assert await store.get_secret("missing") is None
|
||
|
||
async def test_set_overwrites_existing_with_redis(self):
|
||
"""Redis 后端：同名 key 写入覆盖旧值。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
await store.set_secret("k", "old")
|
||
await store.set_secret("k", "new")
|
||
assert await store.get_secret("k") == "new"
|
||
|
||
async def test_delete_secret_with_redis(self):
|
||
"""Redis 后端：删除凭证后不可读取。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
await store.set_secret("k", "v")
|
||
assert await store.delete_secret("k") is True
|
||
assert await store.get_secret("k") is None
|
||
|
||
async def test_delete_nonexistent_returns_false_with_redis(self):
|
||
"""Redis 后端：删除不存在的 key 返回 False。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
assert await store.delete_secret("missing") is False
|
||
|
||
async def test_list_keys_with_redis(self):
|
||
"""Redis 后端：列出全部凭证键。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
await store.set_secret("feishu:a", "1")
|
||
await store.set_secret("feishu:b", "2")
|
||
await store.set_secret("dingtalk:c", "3")
|
||
keys = await store.list_keys()
|
||
assert set(keys) == {"feishu:a", "feishu:b", "dingtalk:c"}
|
||
|
||
async def test_list_keys_with_prefix_with_redis(self):
|
||
"""Redis 后端：按前缀过滤凭证键。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
await store.set_secret("feishu:a", "1")
|
||
await store.set_secret("feishu:b", "2")
|
||
await store.set_secret("dingtalk:c", "3")
|
||
keys = await store.list_keys(prefix="feishu:")
|
||
assert set(keys) == {"feishu:a", "feishu:b"}
|
||
|
||
|
||
# ── Redis=None 降级模式 ─────────────────────────────────
|
||
|
||
|
||
class TestFallbackMode:
|
||
"""redis=None 时降级到内存字典。"""
|
||
|
||
async def test_fallback_set_and_get(self):
|
||
"""降级模式：写入后读取返回明文。"""
|
||
store = SecretsStore(master_key=_make_master_key(), redis=None)
|
||
await store.set_secret("k", "v")
|
||
assert await store.get_secret("k") == "v"
|
||
|
||
async def test_fallback_delete(self):
|
||
"""降级模式：删除凭证后不可读取。"""
|
||
store = SecretsStore(master_key=_make_master_key(), redis=None)
|
||
await store.set_secret("k", "v")
|
||
assert await store.delete_secret("k") is True
|
||
assert await store.get_secret("k") is None
|
||
|
||
async def test_fallback_list_keys_with_prefix(self):
|
||
"""降级模式：按前缀过滤。"""
|
||
store = SecretsStore(master_key=_make_master_key(), redis=None)
|
||
await store.set_secret("feishu:a", "1")
|
||
await store.set_secret("dingtalk:b", "2")
|
||
keys = await store.list_keys(prefix="feishu:")
|
||
assert keys == ["feishu:a"]
|
||
|
||
|
||
# ── 加密往返 ─────────────────────────────────────────────
|
||
|
||
|
||
class TestRedisEncryptionRoundtrip:
|
||
"""Redis 后端的加密-存储-读取-解密往返。"""
|
||
|
||
async def test_stored_value_is_encrypted_in_redis(self):
|
||
"""Redis 中存储的 value 字段为密文，非明文。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
plaintext = "plaintext-token"
|
||
await store.set_secret("k", plaintext)
|
||
# 直接检查 Redis 中的原始值
|
||
raw = redis._data["agentkit:secrets:k"]
|
||
assert plaintext not in raw
|
||
# base64 解码后的密文也不含明文
|
||
import json
|
||
|
||
entry = json.loads(raw)
|
||
ciphertext_bytes = base64.b64decode(entry["value"])
|
||
assert plaintext.encode() not in ciphertext_bytes
|
||
|
||
async def test_roundtrip_preserves_plaintext(self):
|
||
"""加密-存储-读取-解密往返保持明文一致。"""
|
||
redis = _FakeRedis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
plaintext = "feishu-app-secret-12345"
|
||
await store.set_secret("key1", plaintext)
|
||
assert await store.get_secret("key1") == plaintext
|
||
|
||
|
||
# ── 多实例共享语义 ───────────────────────────────────────
|
||
|
||
|
||
class TestMultiInstanceSharing:
|
||
"""两个 store 共用同一 Redis → 多 worker 共享语义。"""
|
||
|
||
async def test_two_stores_share_state_via_redis(self):
|
||
"""worker A set_secret → worker B get_secret 能读到（共享 Redis）。"""
|
||
redis = _FakeRedis()
|
||
store_a = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
store_b = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
|
||
await store_a.set_secret("shared:key", "value-from-a")
|
||
# worker B 通过同一 Redis 能读到 worker A 写入的凭证
|
||
assert await store_b.get_secret("shared:key") == "value-from-a"
|
||
|
||
async def test_delete_from_one_store_visible_to_other(self):
|
||
"""worker A delete → worker B get 返回 None。"""
|
||
redis = _FakeRedis()
|
||
store_a = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
store_b = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
|
||
await store_a.set_secret("k", "v")
|
||
assert await store_b.delete_secret("k") is True
|
||
assert await store_a.get_secret("k") is None
|
||
|
||
|
||
# ── Redis 连接异常 ───────────────────────────────────────
|
||
|
||
|
||
class TestRedisConnectionFailure:
|
||
"""Redis 连接失败时的 fail-closed 行为。"""
|
||
|
||
async def test_get_secret_raises_on_redis_failure(self):
|
||
"""Redis 异常时 get_secret 抛异常（fail-closed，不静默返回 None）。"""
|
||
redis = _make_failing_redis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
with pytest.raises(ConnectionError):
|
||
await store.get_secret("k")
|
||
|
||
async def test_set_secret_raises_on_redis_failure(self):
|
||
"""Redis 异常时 set_secret 抛异常（写入失败不静默降级）。"""
|
||
redis = _make_failing_redis()
|
||
store = SecretsStore(master_key=_make_master_key(), redis=redis)
|
||
with pytest.raises(ConnectionError):
|
||
await store.set_secret("k", "v")
|