7478dc0254
Phase 6.4 - Modular Architecture: - Add lewooogo-brain adapters for LLM providers - Add lewooogo-data dual memory (Redis + PostgreSQL) - Implement consensus engine for multi-agent decisions - Add incident memory service for historical context Phase 9 - Agent Teams (Claude Agent SDK): - Add base agent class with Claude Sonnet 4 integration - Implement action planner, blast radius, and security agents - Add agent API endpoints and proposal workflow - Integrate ADR-009 OpenClaw Agent Teams architecture DevOps & CI/CD: - Add GitHub Actions CI/CD workflows (ci.yaml, cd.yaml) - Add pre-commit hooks and secrets baseline - Add docker-compose for local development - Update Kubernetes network policies Frontend Improvements: - Add auto-healing error boundary component - Update i18n messages for agent features - Enhance dual-state incident card with execution feedback Documentation: - Add 7 ADRs covering MCP, design system, architecture decisions - Update ARCHITECTURE_MEMORY.md with modular design - Add GLOBAL_RULES.md and SOUL.md for project identity Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
402 lines
12 KiB
Python
402 lines
12 KiB
Python
"""
|
||
RedisMemoryProvider 單元測試
|
||
==============================
|
||
|
||
測試案例:
|
||
- test_save_and_load: 儲存並讀取
|
||
- test_delete: 刪除
|
||
- test_ttl_expiry: TTL 相關 (mock)
|
||
- test_key_prefix: Key 前綴
|
||
- test_exists: 存在檢查
|
||
- test_update: 部分更新
|
||
- test_graceful_degradation: 優雅降級
|
||
"""
|
||
|
||
import pytest
|
||
from unittest.mock import patch, AsyncMock
|
||
from pydantic import BaseModel
|
||
|
||
from lewooogo_data.providers.redis_memory import (
|
||
RedisMemoryProvider,
|
||
DEFAULT_TTL_SECONDS,
|
||
)
|
||
|
||
|
||
class SampleModel(BaseModel):
|
||
"""測試用的 Pydantic Model"""
|
||
id: str
|
||
name: str
|
||
value: int = 0
|
||
|
||
|
||
# =============================================================================
|
||
# 基本功能測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderBasic:
|
||
"""基本功能測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_and_load(self, patch_redis_pool, mock_redis):
|
||
"""測試儲存並讀取資料"""
|
||
# Arrange
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="item-1", name="Test Item", value=42)
|
||
|
||
# Act
|
||
save_result = await provider.save("item-1", data)
|
||
|
||
# 驗證儲存成功
|
||
assert save_result is True
|
||
|
||
# 驗證資料已存入 mock storage
|
||
full_key = "test:item-1"
|
||
assert full_key in mock_redis._storage
|
||
|
||
# 載入並驗證
|
||
loaded = await provider.load("item-1")
|
||
|
||
# Assert
|
||
assert loaded is not None
|
||
assert loaded.id == "item-1"
|
||
assert loaded.name == "Test Item"
|
||
assert loaded.value == 42
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete(self, patch_redis_pool, mock_redis):
|
||
"""測試刪除資料"""
|
||
# Arrange
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="item-del", name="To Delete", value=0)
|
||
|
||
# 先儲存
|
||
await provider.save("item-del", data)
|
||
assert "test:item-del" in mock_redis._storage
|
||
|
||
# Act
|
||
delete_result = await provider.delete("item-del")
|
||
|
||
# Assert
|
||
assert delete_result is True
|
||
assert "test:item-del" not in mock_redis._storage
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_nonexistent(self, patch_redis_pool, mock_redis):
|
||
"""測試刪除不存在的資料"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
delete_result = await provider.delete("nonexistent")
|
||
|
||
# Assert
|
||
assert delete_result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_load_nonexistent(self, patch_redis_pool, mock_redis):
|
||
"""測試載入不存在的資料"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
loaded = await provider.load("nonexistent")
|
||
|
||
# Assert
|
||
assert loaded is None
|
||
|
||
|
||
# =============================================================================
|
||
# TTL 測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderTTL:
|
||
"""TTL 相關測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_ttl(self, patch_redis_pool, mock_redis):
|
||
"""測試預設 TTL (7 天)"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="ttl-1", name="TTL Test", value=1)
|
||
|
||
# Act
|
||
await provider.save("ttl-1", data)
|
||
|
||
# Assert - 驗證 TTL 為預設值
|
||
assert mock_redis._ttls["test:ttl-1"] == DEFAULT_TTL_SECONDS
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_custom_ttl(self, patch_redis_pool, mock_redis):
|
||
"""測試自訂 TTL"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="ttl-2", name="Custom TTL", value=2)
|
||
|
||
# Act - 設定 1 小時 TTL
|
||
custom_ttl = 3600
|
||
await provider.save("ttl-2", data, ttl_seconds=custom_ttl)
|
||
|
||
# Assert
|
||
assert mock_redis._ttls["test:ttl-2"] == custom_ttl
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_ttl(self, patch_redis_pool, mock_redis):
|
||
"""測試取得剩餘 TTL"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="ttl-3", name="Get TTL", value=3)
|
||
|
||
await provider.save("ttl-3", data, ttl_seconds=1800)
|
||
|
||
# Act
|
||
ttl = await provider.get_ttl("ttl-3")
|
||
|
||
# Assert
|
||
assert ttl == 1800
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_ttl_nonexistent(self, patch_redis_pool, mock_redis):
|
||
"""測試取得不存在 key 的 TTL"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
ttl = await provider.get_ttl("nonexistent")
|
||
|
||
# Assert
|
||
assert ttl == -2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_extend_ttl(self, patch_redis_pool, mock_redis):
|
||
"""測試延長 TTL"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="ttl-4", name="Extend TTL", value=4)
|
||
|
||
await provider.save("ttl-4", data, ttl_seconds=1000)
|
||
|
||
# Act - 延長 500 秒
|
||
result = await provider.extend_ttl("ttl-4", 500)
|
||
|
||
# Assert
|
||
assert result is True
|
||
assert mock_redis._ttls["test:ttl-4"] == 1500
|
||
|
||
|
||
# =============================================================================
|
||
# Key Prefix 測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderKeyPrefix:
|
||
"""Key 前綴測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_key_prefix(self, patch_redis_pool, mock_redis):
|
||
"""測試 Key 前綴正確套用"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="myapp:incidents",
|
||
)
|
||
data = SampleModel(id="inc-1", name="Incident", value=100)
|
||
|
||
# Act
|
||
await provider.save("inc-1", data)
|
||
|
||
# Assert
|
||
expected_key = "myapp:incidents:inc-1"
|
||
assert expected_key in mock_redis._storage
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_different_prefixes_isolated(self, patch_redis_pool, mock_redis):
|
||
"""測試不同前綴的資料隔離"""
|
||
provider_a = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="app-a",
|
||
)
|
||
provider_b = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="app-b",
|
||
)
|
||
|
||
data_a = SampleModel(id="shared-key", name="App A", value=1)
|
||
data_b = SampleModel(id="shared-key", name="App B", value=2)
|
||
|
||
# Act
|
||
await provider_a.save("shared-key", data_a)
|
||
await provider_b.save("shared-key", data_b)
|
||
|
||
# Assert - 兩個 provider 的資料分開
|
||
assert "app-a:shared-key" in mock_redis._storage
|
||
assert "app-b:shared-key" in mock_redis._storage
|
||
|
||
loaded_a = await provider_a.load("shared-key")
|
||
loaded_b = await provider_b.load("shared-key")
|
||
|
||
assert loaded_a.name == "App A"
|
||
assert loaded_b.name == "App B"
|
||
|
||
|
||
# =============================================================================
|
||
# 存在檢查測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderExists:
|
||
"""存在檢查測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_exists_true(self, patch_redis_pool, mock_redis):
|
||
"""測試存在的 key"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="exist-1", name="Exists", value=1)
|
||
await provider.save("exist-1", data)
|
||
|
||
# Act
|
||
result = await provider.exists("exist-1")
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_exists_false(self, patch_redis_pool, mock_redis):
|
||
"""測試不存在的 key"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
result = await provider.exists("nonexistent")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
|
||
# =============================================================================
|
||
# 部分更新測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderUpdate:
|
||
"""部分更新測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update(self, patch_redis_pool, mock_redis):
|
||
"""測試部分更新"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="upd-1", name="Original", value=10)
|
||
await provider.save("upd-1", data, ttl_seconds=3600)
|
||
|
||
# Act - 只更新 value
|
||
result = await provider.update("upd-1", {"value": 20})
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
loaded = await provider.load("upd-1")
|
||
assert loaded.name == "Original" # 保持不變
|
||
assert loaded.value == 20 # 已更新
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_nonexistent(self, patch_redis_pool, mock_redis):
|
||
"""測試更新不存在的資料"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
result = await provider.update("nonexistent", {"value": 999})
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
|
||
# =============================================================================
|
||
# 優雅降級測試
|
||
# =============================================================================
|
||
|
||
class TestRedisMemoryProviderGracefulDegradation:
|
||
"""優雅降級測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_load_without_init(self):
|
||
"""測試未初始化時 load 回傳 None"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act - 沒有 patch_redis_pool，pool 未初始化
|
||
result = await provider.load("any-key")
|
||
|
||
# Assert - 優雅降級，返回 None 而非拋異常
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_without_init(self):
|
||
"""測試未初始化時 save 回傳 False"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
data = SampleModel(id="x", name="X", value=0)
|
||
|
||
# Act
|
||
result = await provider.save("x", data)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_without_init(self):
|
||
"""測試未初始化時 delete 回傳 False"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
result = await provider.delete("any-key")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_exists_without_init(self):
|
||
"""測試未初始化時 exists 回傳 False"""
|
||
provider = RedisMemoryProvider(
|
||
model_class=SampleModel,
|
||
key_prefix="test",
|
||
)
|
||
|
||
# Act
|
||
result = await provider.exists("any-key")
|
||
|
||
# Assert
|
||
assert result is False
|