6f049877fc
- Python: ruff --fix 修復 280 個 lint 錯誤 - lewooogo-core: src/ 目錄未追蹤，導致 CI eslint 失敗 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
463 lines
14 KiB
Python
463 lines
14 KiB
Python
"""
|
||
Multi-Sig Redis 自動化測試腳本
|
||
==============================
|
||
Phase 6.1.1: 全自動單元自檢
|
||
|
||
測試項目:
|
||
1. Redis 連線池初始化
|
||
2. 簽核單 CRUD 操作
|
||
3. 分散式鎖競爭測試
|
||
4. TTL 驗證 (7 天)
|
||
5. 雙重簽核防禦
|
||
|
||
統帥鐵律:
|
||
- 禁止人工 QA，此腳本必須全自動執行
|
||
- 輸出必須為 Raw Data (stdout logs)
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
from datetime import UTC, datetime
|
||
from uuid import uuid4
|
||
|
||
# 添加專案路徑
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||
|
||
import structlog
|
||
|
||
# 配置 structlog 輸出
|
||
structlog.configure(
|
||
processors=[
|
||
structlog.processors.TimeStamper(fmt="iso"),
|
||
structlog.dev.ConsoleRenderer(),
|
||
],
|
||
wrapper_class=structlog.make_filtering_bound_logger(0),
|
||
)
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
|
||
async def test_redis_connection():
|
||
"""測試 1: Redis 連線池初始化"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_1_REDIS_CONNECTION", status="starting")
|
||
|
||
from src.core.redis_client import get_redis, init_redis_pool
|
||
|
||
try:
|
||
# 初始化連線池
|
||
pool = await init_redis_pool()
|
||
logger.info("redis_pool_initialized", pool_type=type(pool).__name__)
|
||
|
||
# 取得連線
|
||
redis_client = get_redis()
|
||
|
||
# PING 測試
|
||
pong = await redis_client.ping()
|
||
logger.info("redis_ping", response=pong)
|
||
|
||
# 寫入測試值
|
||
test_key = "test:connection:check"
|
||
await redis_client.set(test_key, "awoooi_phase6", ex=60)
|
||
value = await redis_client.get(test_key)
|
||
logger.info("redis_set_get", key=test_key, value=value)
|
||
|
||
# 清理測試值
|
||
await redis_client.delete(test_key)
|
||
|
||
logger.info("TEST_1_REDIS_CONNECTION", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_1_REDIS_CONNECTION", status="FAILED", error=str(e))
|
||
return False
|
||
|
||
|
||
async def test_approval_crud():
|
||
"""測試 2: 簽核單 CRUD 操作"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_2_APPROVAL_CRUD", status="starting")
|
||
|
||
from src.services.multi_sig_redis import get_multi_sig_redis_service
|
||
|
||
service = get_multi_sig_redis_service()
|
||
approval_id = str(uuid4())
|
||
|
||
try:
|
||
# CREATE
|
||
state = await service.create_approval(
|
||
approval_id=approval_id,
|
||
action="DELETE_POD",
|
||
description="測試簽核單 - Phase 6.1.1 自動化測試",
|
||
risk_level="high",
|
||
required_signatures=2,
|
||
namespace="awoooi",
|
||
resource_name="test-pod-001",
|
||
)
|
||
logger.info("approval_created",
|
||
id=state["id"],
|
||
status=state["status"],
|
||
required=state["required_signatures"])
|
||
|
||
# READ
|
||
retrieved = await service.get_approval(approval_id)
|
||
assert retrieved is not None, "Approval not found after create"
|
||
assert retrieved["status"] == "pending", f"Expected pending, got {retrieved['status']}"
|
||
logger.info("approval_retrieved",
|
||
id=retrieved["id"],
|
||
signatures_count=len(retrieved["signatures"]))
|
||
|
||
# EXISTS CHECK
|
||
exists = await service.exists(approval_id)
|
||
assert exists, "Approval should exist"
|
||
logger.info("approval_exists", exists=exists)
|
||
|
||
# UPDATE (reject)
|
||
rejected = await service.reject_approval(
|
||
approval_id=approval_id,
|
||
rejector_id="test-ciso",
|
||
rejector_name="資安長測試",
|
||
reason="Phase 6.1.1 自動化測試拒絕",
|
||
)
|
||
assert rejected["status"] == "rejected", f"Expected rejected, got {rejected['status']}"
|
||
logger.info("approval_rejected",
|
||
status=rejected["status"],
|
||
rejector=rejected.get("rejector_name"))
|
||
|
||
logger.info("TEST_2_APPROVAL_CRUD", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_2_APPROVAL_CRUD", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_signature_flow():
|
||
"""測試 3: 簽核流程 (含分散式鎖)"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_3_SIGNATURE_FLOW", status="starting")
|
||
|
||
from src.services.multi_sig_redis import get_multi_sig_redis_service
|
||
|
||
service = get_multi_sig_redis_service()
|
||
approval_id = str(uuid4())
|
||
|
||
try:
|
||
# 建立需要 2 人簽核的單子
|
||
await service.create_approval(
|
||
approval_id=approval_id,
|
||
action="RESTART_SERVICE",
|
||
description="測試簽核流程",
|
||
risk_level="critical",
|
||
required_signatures=2,
|
||
namespace="awoooi",
|
||
)
|
||
logger.info("approval_created_for_signing", id=approval_id, required=2)
|
||
|
||
# 第一人簽核
|
||
state1 = await service.add_signature(
|
||
approval_id=approval_id,
|
||
signer_id="cto-001",
|
||
signer_name="技術長",
|
||
comment="同意執行",
|
||
source="web",
|
||
)
|
||
logger.info("signature_1_added",
|
||
current=state1["current_signatures"],
|
||
required=state1["required_signatures"],
|
||
status=state1["status"])
|
||
assert state1["status"] == "pending", "Should still be pending with 1/2 signatures"
|
||
|
||
# 第二人簽核 (應該觸發 approved)
|
||
state2 = await service.add_signature(
|
||
approval_id=approval_id,
|
||
signer_id="ceo-001",
|
||
signer_name="執行長",
|
||
comment="核准",
|
||
source="telegram",
|
||
telegram_user_id=123456789,
|
||
)
|
||
logger.info("signature_2_added",
|
||
current=state2["current_signatures"],
|
||
required=state2["required_signatures"],
|
||
status=state2["status"])
|
||
assert state2["status"] == "approved", f"Should be approved, got {state2['status']}"
|
||
|
||
logger.info("TEST_3_SIGNATURE_FLOW", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_3_SIGNATURE_FLOW", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_duplicate_signature_defense():
|
||
"""測試 4: 雙重簽核防禦"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_4_DUPLICATE_SIGNATURE_DEFENSE", status="starting")
|
||
|
||
from src.services.multi_sig_redis import get_multi_sig_redis_service
|
||
|
||
service = get_multi_sig_redis_service()
|
||
approval_id = str(uuid4())
|
||
|
||
try:
|
||
await service.create_approval(
|
||
approval_id=approval_id,
|
||
action="SCALE_DEPLOYMENT",
|
||
description="雙重簽核防禦測試",
|
||
risk_level="medium",
|
||
required_signatures=3,
|
||
)
|
||
|
||
# 第一次簽核
|
||
await service.add_signature(
|
||
approval_id=approval_id,
|
||
signer_id="same-user",
|
||
signer_name="測試用戶",
|
||
)
|
||
logger.info("first_signature_success", signer="same-user")
|
||
|
||
# 嘗試重複簽核 (應該被拒絕)
|
||
try:
|
||
await service.add_signature(
|
||
approval_id=approval_id,
|
||
signer_id="same-user",
|
||
signer_name="測試用戶",
|
||
)
|
||
logger.error("duplicate_signature_allowed", status="SECURITY_BREACH")
|
||
return False
|
||
except RuntimeError as e:
|
||
if "Already signed" in str(e):
|
||
logger.info("duplicate_signature_blocked", error=str(e))
|
||
else:
|
||
raise
|
||
|
||
logger.info("TEST_4_DUPLICATE_SIGNATURE_DEFENSE", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_4_DUPLICATE_SIGNATURE_DEFENSE", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_ttl_verification():
|
||
"""測試 5: TTL 驗證 (7 天 = 604800 秒)"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_5_TTL_VERIFICATION", status="starting")
|
||
|
||
from src.core.redis_client import get_redis
|
||
from src.services.multi_sig_redis import (
|
||
APPROVAL_TTL_SECONDS,
|
||
get_multi_sig_redis_service,
|
||
)
|
||
|
||
service = get_multi_sig_redis_service()
|
||
redis_client = get_redis()
|
||
approval_id = str(uuid4())
|
||
|
||
try:
|
||
await service.create_approval(
|
||
approval_id=approval_id,
|
||
action="TTL_TEST",
|
||
description="TTL 驗證測試",
|
||
risk_level="low",
|
||
required_signatures=1,
|
||
)
|
||
|
||
# 檢查 TTL
|
||
key = f"approval:{approval_id}"
|
||
ttl = await redis_client.ttl(key)
|
||
|
||
logger.info("ttl_check",
|
||
key=key,
|
||
ttl_seconds=ttl,
|
||
expected_ttl=APPROVAL_TTL_SECONDS,
|
||
ttl_days=ttl / 86400 if ttl > 0 else 0)
|
||
|
||
# TTL 應該接近 604800 秒 (允許 10 秒誤差)
|
||
assert ttl > APPROVAL_TTL_SECONDS - 10, f"TTL too low: {ttl}"
|
||
assert ttl <= APPROVAL_TTL_SECONDS, f"TTL too high: {ttl}"
|
||
|
||
logger.info("TEST_5_TTL_VERIFICATION", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_5_TTL_VERIFICATION", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_concurrent_signatures():
|
||
"""測試 6: 併發簽核測試 (分散式鎖壓力測試)"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_6_CONCURRENT_SIGNATURES", status="starting")
|
||
|
||
from src.services.multi_sig_redis import get_multi_sig_redis_service
|
||
|
||
service = get_multi_sig_redis_service()
|
||
approval_id = str(uuid4())
|
||
|
||
try:
|
||
await service.create_approval(
|
||
approval_id=approval_id,
|
||
action="CONCURRENT_TEST",
|
||
description="併發鎖測試",
|
||
risk_level="high",
|
||
required_signatures=5,
|
||
)
|
||
|
||
# 模擬 5 個不同用戶同時簽核
|
||
async def sign(user_num: int):
|
||
try:
|
||
result = await service.add_signature(
|
||
approval_id=approval_id,
|
||
signer_id=f"user-{user_num}",
|
||
signer_name=f"用戶 {user_num}",
|
||
source="concurrent_test",
|
||
)
|
||
return ("success", user_num, result["current_signatures"])
|
||
except Exception as e:
|
||
return ("error", user_num, str(e))
|
||
|
||
# 同時發起 5 個簽核請求
|
||
tasks = [sign(i) for i in range(1, 6)]
|
||
results = await asyncio.gather(*tasks)
|
||
|
||
success_count = sum(1 for r in results if r[0] == "success")
|
||
error_count = sum(1 for r in results if r[0] == "error")
|
||
|
||
for status, user_num, detail in results:
|
||
logger.info("concurrent_result",
|
||
user=user_num,
|
||
status=status,
|
||
detail=detail)
|
||
|
||
logger.info("concurrent_summary",
|
||
success=success_count,
|
||
errors=error_count)
|
||
|
||
# 驗證最終狀態
|
||
final = await service.get_approval(approval_id)
|
||
logger.info("final_state",
|
||
current_signatures=final["current_signatures"],
|
||
status=final["status"])
|
||
|
||
# 所有 5 個簽核都應成功
|
||
assert success_count == 5, f"Expected 5 successes, got {success_count}"
|
||
assert final["status"] == "approved", f"Expected approved, got {final['status']}"
|
||
|
||
logger.info("TEST_6_CONCURRENT_SIGNATURES", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_6_CONCURRENT_SIGNATURES", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_list_pending():
|
||
"""測試 7: 列出待簽核單"""
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_7_LIST_PENDING", status="starting")
|
||
|
||
from src.services.multi_sig_redis import get_multi_sig_redis_service
|
||
|
||
service = get_multi_sig_redis_service()
|
||
|
||
try:
|
||
# 建立幾個待簽核單
|
||
ids = []
|
||
for i in range(3):
|
||
approval_id = str(uuid4())
|
||
await service.create_approval(
|
||
approval_id=approval_id,
|
||
action=f"LIST_TEST_{i}",
|
||
description=f"列表測試 {i}",
|
||
risk_level="low",
|
||
required_signatures=1,
|
||
)
|
||
ids.append(approval_id)
|
||
|
||
# 列出待簽核單
|
||
pending = await service.list_pending(limit=100)
|
||
logger.info("pending_list_count", count=len(pending))
|
||
|
||
# 應該至少包含我們建立的 3 個
|
||
found = sum(1 for p in pending if p["id"] in ids)
|
||
logger.info("found_our_approvals", found=found, expected=3)
|
||
|
||
assert found >= 3, f"Expected at least 3, found {found}"
|
||
|
||
logger.info("TEST_7_LIST_PENDING", status="PASSED")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error("TEST_7_LIST_PENDING", status="FAILED", error=str(e))
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def main():
|
||
"""主測試入口"""
|
||
logger.info("=" * 60)
|
||
logger.info("PHASE_6_1_1_REDIS_MULTISIG_TEST", status="STARTING")
|
||
logger.info("timestamp", time=datetime.now(UTC).isoformat())
|
||
logger.info("=" * 60)
|
||
|
||
results = {}
|
||
|
||
# 測試 1: Redis 連線
|
||
results["redis_connection"] = await test_redis_connection()
|
||
|
||
if not results["redis_connection"]:
|
||
logger.error("CRITICAL", message="Redis 連線失敗，終止測試")
|
||
return
|
||
|
||
# 測試 2-7
|
||
results["approval_crud"] = await test_approval_crud()
|
||
results["signature_flow"] = await test_signature_flow()
|
||
results["duplicate_defense"] = await test_duplicate_signature_defense()
|
||
results["ttl_verification"] = await test_ttl_verification()
|
||
results["concurrent_signatures"] = await test_concurrent_signatures()
|
||
results["list_pending"] = await test_list_pending()
|
||
|
||
# 關閉連線池
|
||
from src.core.redis_client import close_redis_pool
|
||
await close_redis_pool()
|
||
|
||
# 總結報告
|
||
logger.info("=" * 60)
|
||
logger.info("TEST_SUMMARY")
|
||
|
||
passed = sum(1 for v in results.values() if v)
|
||
failed = sum(1 for v in results.values() if not v)
|
||
|
||
for test_name, passed_flag in results.items():
|
||
status = "✅ PASSED" if passed_flag else "❌ FAILED"
|
||
logger.info(f" {test_name}: {status}")
|
||
|
||
logger.info("-" * 60)
|
||
logger.info(f"TOTAL: {passed} passed, {failed} failed")
|
||
logger.info("=" * 60)
|
||
|
||
if failed > 0:
|
||
sys.exit(1)
|
||
else:
|
||
logger.info("ALL_TESTS_PASSED", message="Phase 6.1.1 Redis Multi-Sig 驗證完成")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|