feat(api): HostRepairAgent — SSH 主機層修復 (Task 11)

- host_repair_agent.py: layer路由、command injection防護、asyncio SSH執行
- 測試: 12 cases 全通過 (routing/sanitize/success/fail/timeout/denied)
- SSH key: /etc/repair-ssh/id_ed25519 (K8s secret mount)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-05 11:22:00 +08:00
parent 892c5d53a7
commit e7d8da85f6
2 changed files with 264 additions and 0 deletions

View File

@@ -0,0 +1,130 @@
"""
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",
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