4b24ecd67f
C1: _ssh_execute 直接接收 key_path 參數，不反查 LAYER_SSH_CONFIG
C2: PlaybookService.create() proxy，Router 不再穿透呼叫 _repository
C3: CD Step 1b sed 替換 IMAGE_TAG_PLACEHOLDER，消除失敗中斷風險
M3: repair-bot 110/188 regex 統一 [a-z0-9][a-z0-9-]{0,30}，禁止底線
m1: defaultMode 0400 加八進位說明注釋
m2: _ssh_execute 用 deadline 計算剩餘 timeout
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
132 lines
5.2 KiB
Python
132 lines
5.2 KiB
Python
"""
|
||
tests/test_host_repair_agent.py
|
||
Host Repair Agent 單元測試
|
||
不需要實際 SSH 連線 — 測試路由邏輯和命令組裝
|
||
"""
|
||
import asyncio
|
||
import pytest
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
|
||
# =============================================================================
|
||
# 測試 HostRepairConfig 路由
|
||
# =============================================================================
|
||
|
||
class TestHostRepairConfig:
|
||
def test_layer_docker_110_routes_to_110(self):
|
||
from src.services.host_repair_agent import get_ssh_config_for_layer
|
||
config = get_ssh_config_for_layer("docker-110")
|
||
assert config["user"] == "wooo"
|
||
assert config["host"] == "192.168.0.110"
|
||
|
||
def test_layer_docker_188_routes_to_188(self):
|
||
from src.services.host_repair_agent import get_ssh_config_for_layer
|
||
config = get_ssh_config_for_layer("docker-188")
|
||
assert config["user"] == "ollama"
|
||
assert config["host"] == "192.168.0.188"
|
||
|
||
def test_layer_systemd_188_routes_to_188(self):
|
||
from src.services.host_repair_agent import get_ssh_config_for_layer
|
||
config = get_ssh_config_for_layer("systemd-188")
|
||
assert config["user"] == "ollama"
|
||
assert config["host"] == "192.168.0.188"
|
||
|
||
def test_unknown_layer_raises(self):
|
||
from src.services.host_repair_agent import get_ssh_config_for_layer
|
||
with pytest.raises(ValueError, match="Unknown layer"):
|
||
get_ssh_config_for_layer("unknown-layer")
|
||
|
||
def test_k8s_layer_raises(self):
|
||
"""k8s layer 不走 SSH，應 raise"""
|
||
from src.services.host_repair_agent import get_ssh_config_for_layer
|
||
with pytest.raises(ValueError, match="kubectl"):
|
||
get_ssh_config_for_layer("k8s")
|
||
|
||
|
||
# =============================================================================
|
||
# 測試 SSH 命令組裝
|
||
# =============================================================================
|
||
|
||
class TestSSHCommandBuilding:
|
||
def test_repair_command_format(self):
|
||
from src.services.host_repair_agent import build_repair_command
|
||
cmd = build_repair_command("sentry")
|
||
assert cmd == "repair:sentry"
|
||
|
||
def test_repair_command_component_sanitized(self):
|
||
"""防止 command injection"""
|
||
from src.services.host_repair_agent import build_repair_command
|
||
with pytest.raises(ValueError, match="Invalid component"):
|
||
build_repair_command("sentry; rm -rf /")
|
||
|
||
def test_repair_command_valid_components(self):
|
||
from src.services.host_repair_agent import build_repair_command
|
||
valid = ["sentry", "harbor", "gitea", "openclaw", "gitea-runner", "alertmanager", "redis", "nginx"]
|
||
for component in valid:
|
||
cmd = build_repair_command(component)
|
||
assert cmd == f"repair:{component}"
|
||
|
||
|
||
# =============================================================================
|
||
# 測試 HostRepairAgent.repair() 路由
|
||
# =============================================================================
|
||
|
||
class TestHostRepairAgent:
|
||
@pytest.mark.asyncio
|
||
async def test_repair_success_returns_ok(self):
|
||
from src.services.host_repair_agent import HostRepairAgent
|
||
|
||
agent = HostRepairAgent()
|
||
with patch.object(agent, "_ssh_execute", new_callable=AsyncMock) as mock_ssh:
|
||
mock_ssh.return_value = "REPAIR_OK:sentry"
|
||
|
||
result = await agent.repair(layer="docker-110", component="sentry")
|
||
|
||
assert result.success is True
|
||
assert result.component == "sentry"
|
||
assert result.layer == "docker-110"
|
||
mock_ssh.assert_called_once_with(
|
||
host="192.168.0.110",
|
||
user="wooo",
|
||
key_path="/etc/repair-ssh/id_ed25519",
|
||
command="repair:sentry"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_repair_fail_returns_failure(self):
|
||
from src.services.host_repair_agent import HostRepairAgent
|
||
|
||
agent = HostRepairAgent()
|
||
with patch.object(agent, "_ssh_execute", new_callable=AsyncMock) as mock_ssh:
|
||
mock_ssh.return_value = "REPAIR_FAIL:harbor:exit_1"
|
||
|
||
result = await agent.repair(layer="docker-110", component="harbor")
|
||
|
||
assert result.success is False
|
||
assert "REPAIR_FAIL" in result.error
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_repair_ssh_timeout_returns_failure(self):
|
||
from src.services.host_repair_agent import HostRepairAgent
|
||
|
||
agent = HostRepairAgent()
|
||
with patch.object(agent, "_ssh_execute", new_callable=AsyncMock) as mock_ssh:
|
||
mock_ssh.side_effect = asyncio.TimeoutError()
|
||
|
||
result = await agent.repair(layer="docker-110", component="sentry")
|
||
|
||
assert result.success is False
|
||
assert "timeout" in result.error.lower()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_repair_denied_returns_failure(self):
|
||
from src.services.host_repair_agent import HostRepairAgent
|
||
|
||
agent = HostRepairAgent()
|
||
with patch.object(agent, "_ssh_execute", new_callable=AsyncMock) as mock_ssh:
|
||
mock_ssh.return_value = "REPAIR_DENIED:unknown_component:badcomponent"
|
||
|
||
result = await agent.repair(layer="docker-110", component="badcomponent")
|
||
|
||
assert result.success is False
|