fix(telegram): separate ssh diagnosis from repair failures
All checks were successful
Code Review / ai-code-review (push) Successful in 11s
CD Pipeline / tests (push) Successful in 1m27s
CD Pipeline / build-and-deploy (push) Successful in 4m19s
CD Pipeline / post-deploy-checks (push) Successful in 1m25s

This commit is contained in:
Your Name
2026-05-06 22:03:19 +08:00
parent 53994e75f0
commit 9dfecc4d1b
5 changed files with 107 additions and 8 deletions

View File

@@ -134,3 +134,37 @@ class TestDockerPruneActionRouting:
target="unknown",
)
assert manager._ssh.execute.call_args.kwargs["tool_name"] == "ssh_diagnose"
@pytest.mark.asyncio
async def test_failed_diagnose_does_not_emit_auto_repair_failed(self, manager, monkeypatch):
"""ssh_diagnose 是只讀診斷,失敗時不可標成自動修復失敗。"""
incident = _fake_incident()
token = _fake_token()
auto_result_calls = []
manager._ssh.execute = AsyncMock(
return_value=MCPToolResult(
success=False,
execution_id="exec-fail",
error="diagnosis connection failed",
)
)
monkeypatch.setattr(
"src.services.decision_manager._push_auto_repair_result",
lambda *args, **kwargs: auto_result_calls.append((args, kwargs)),
)
monkeypatch.setattr(
"src.services.decision_manager._escalate_decision_auto_repair_unavailable",
lambda *args, **kwargs: None,
)
await manager._ssh_execute(
incident=incident,
token=token,
action="ssh 192.168.0.110 'df -h && free -h'",
target="unknown",
)
assert manager._ssh.execute.call_args.kwargs["tool_name"] == "ssh_diagnose"
assert token.proposal_data["automation_state"] == "diagnosis_failed_manual_required"
assert auto_result_calls == []

View File

@@ -15,8 +15,8 @@ from src.services.telegram_gateway import (
RepairReportMessage,
ResourceWarnMessage,
SentryErrorMessage,
TelegramMessage,
TelegramGateway,
TelegramMessage,
)
@@ -84,6 +84,42 @@ class TestTelegramMessageFormat:
assert "處置狀態" in result
assert "AI 無可安全執行動作,需人工判斷" in result
def test_telegram_message_diagnosis_state_is_not_auto_repair(self):
"""SSH 只讀診斷 lane 不得被顯示成自動修復。"""
msg = TelegramMessage(
status_emoji="⚠️",
risk_level="MEDIUM",
resource_name="node-110",
root_cause="SSH diagnosis collected",
suggested_action="ssh 192.168.0.110 'uptime'",
estimated_downtime="unknown",
approval_id="INC-20260506-DIAG",
automation_state="diagnosis_collected_manual_required",
)
result = msg.format()
assert "AI 已完成只讀診斷,需人工判斷" in result
assert "AI 自動修復失敗" not in result
def test_telegram_message_diagnosis_failure_state(self):
"""SSH 診斷工具失敗必須標成診斷失敗,而不是修復失敗。"""
msg = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="node-110",
root_cause="SSH MCP execution failed",
suggested_action="ssh 192.168.0.110 'uptime'",
estimated_downtime="unknown",
approval_id="INC-20260506-DIAGFAIL",
automation_state="diagnosis_failed_manual_required",
)
result = msg.format()
assert "AI 診斷工具失敗,需人工排查" in result
assert "AI 自動修復失敗" not in result
def test_telegram_message_with_token_cost(self):
"""測試含 Token/Cost 的訊息"""
msg = TelegramMessage(