feat(adr-076): Task 3.3 — SSH 修復 KM 萃取(補齊飛輪雙手)
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled

動機: SSH MCP 修復(docker restart/systemctl)成功後,KM 無法學習
因為 _extract_repair_steps 只處理 kubectl,SSH 路徑完全漏失。

approval_execution.py:
  - _trigger_playbook_extraction: 成功執行後將 approval.action 寫入
    incident.outcome.learning_notes,供 Playbook 萃取器讀取

playbook_service.py:
  - _parse_ssh_command(): 新增模組函式,解析 ssh [user@]host 'cmd' 格式
  - _extract_repair_steps(): 步驟 2 擴充 SSH 路徑分支
      ssh ... → ActionType.SSH_COMMAND + host 記錄
      kubectl ... → ActionType.KUBECTL(保留原有邏輯)
  - _generate_name(): SSH 修復自動加 [SSH] 前綴
  - _extract_tags(): SSH 修復自動加 ssh + host_layer 標籤

test_playbook_ssh_extraction.py: 18 tests(100% 通過)

飛輪雙手對齊:
  kubectl 路徑: decision_chain.reasoning_steps → KM  (既有)
  SSH 路徑: approval.action → learning_notes → KM  (Task 3.3 新增)

測試: 794 passed, 26 skipped, 0 failed

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-14 15:19:54 +08:00
parent cc42aa0bdb
commit aae7c12645
3 changed files with 389 additions and 14 deletions

View File

@@ -0,0 +1,295 @@
"""
Playbook SSH 修復 KM 萃取測試
=================================
Task 3.3: SSH 修復路徑 Playbook 萃取
測試範圍:
- _parse_ssh_command() — 各種 SSH 格式解析
- PlaybookService._extract_repair_steps() — SSH / kubectl 路徑
- PlaybookService._generate_name() — [SSH] 前綴
- PlaybookService._extract_tags() — ssh / host_layer 標籤
🔴 遵循「禁止 Mock 測試鐵律」
- 純 Python 邏輯:不需要 DB/Redis/Telegram
- 使用真實模型物件,不 Mock
建立: 2026-04-14 (台北時區) Claude Sonnet 4.6 (Task 3.3)
"""
from datetime import datetime, timezone
import pytest
from src.models.incident import AIDecisionChain, Incident, IncidentOutcome, Severity, Signal
from src.models.playbook import ActionType, RiskLevel
from src.services.playbook_service import PlaybookService, _parse_ssh_command
_TZ_TAIPEI = timezone.utc # 測試用 UTC不影響邏輯
def _make_incident(**kwargs) -> Incident:
"""建立最小化 Incident純記憶體"""
now = datetime.now(_TZ_TAIPEI)
signal = Signal(
alert_name=kwargs.pop("alert_name", "MinioDown"),
severity=Severity.P2,
source="alertmanager",
fired_at=now,
labels={},
annotations={},
)
return Incident(
incident_id=kwargs.pop("incident_id", "INC-20260414-TEST"),
title=kwargs.pop("title", "Test Incident"),
severity=Severity.P2,
signals=[signal],
affected_services=kwargs.pop("affected_services", ["minio"]),
proposal_ids=[],
**kwargs,
)
# =============================================================================
# _parse_ssh_command
# =============================================================================
class TestParseSshCommand:
"""SSH 指令解析邏輯"""
@pytest.mark.parametrize("cmd,expected_host,expected_inner", [
(
"ssh 192.168.0.188 'docker restart minio'",
"192.168.0.188",
"docker restart minio",
),
(
"ssh root@192.168.0.110 'systemctl restart ollama || docker restart ollama'",
"192.168.0.110",
"systemctl restart ollama || docker restart ollama",
),
(
"ssh {host} \"cd /data/harbor && docker-compose up -d\"",
"{host}",
"cd /data/harbor && docker-compose up -d",
),
])
def test_parse_standard_ssh(self, cmd, expected_host, expected_inner):
"""標準 SSH 格式解析"""
host, inner = _parse_ssh_command(cmd)
assert host == expected_host
assert inner == expected_inner
def test_parse_unrecognized_returns_empty_host(self):
"""無法解析的格式 → 空 host保留原始命令"""
cmd = "ssh host_without_quotes do_something"
host, inner = _parse_ssh_command(cmd)
assert host == ""
assert inner == cmd
def test_parse_empty_string(self):
"""空字串不崩潰"""
host, inner = _parse_ssh_command("")
assert host == ""
# =============================================================================
# _extract_repair_steps — SSH path (last_repair_action)
# =============================================================================
class TestExtractRepairStepsSSH:
"""SSH 修復路徑萃取Task 3.3 新增)"""
def _svc(self) -> PlaybookService:
return PlaybookService.__new__(PlaybookService) # 不觸發 __init__
def test_ssh_command_in_last_repair_action(self):
"""last_repair_action 含 SSH → ActionType.SSH_COMMAND"""
incident = _make_incident()
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh 192.168.0.188 'docker restart minio'"
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert len(steps) == 1
assert steps[0].action_type == ActionType.SSH_COMMAND
assert "docker restart minio" in steps[0].command
def test_ssh_fallback_when_no_decision_chain(self):
"""無 decision_chain 時SSH last_repair_action 能補位"""
incident = _make_incident()
incident.decision_chain = None
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh root@192.168.0.110 'systemctl restart ollama'"
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert steps
assert steps[0].action_type == ActionType.SSH_COMMAND
def test_kubectl_in_last_repair_action(self):
"""last_repair_action 含 kubectl → ActionType.KUBECTL"""
incident = _make_incident()
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = (
"kubectl rollout restart deployment/awoooi-api -n awoooi-prod"
)
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert len(steps) == 1
assert steps[0].action_type == ActionType.KUBECTL
assert "kubectl rollout restart" in steps[0].command
def test_decision_chain_takes_priority_over_learning_notes(self):
"""decision_chain.reasoning_steps 有 kubectl → learning_notes SSH 不覆蓋"""
now = datetime.now(_TZ_TAIPEI)
incident = _make_incident()
incident.decision_chain = AIDecisionChain(
model_used="deepseek-r1:14b",
hypothesis="Pod crash",
confidence=0.85,
reasoning_steps=["kubectl rollout restart deployment/test -n prod"],
inference_started_at=now,
inference_completed_at=now,
latency_ms=100,
)
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh 192.168.0.188 'docker restart minio'"
svc = self._svc()
steps = svc._extract_repair_steps(incident)
# decision_chain 優先
assert steps[0].action_type == ActionType.KUBECTL
def test_no_action_returns_empty(self):
"""無任何修復來源 → 空列表"""
incident = _make_incident()
incident.decision_chain = None
incident.outcome = None
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert steps == []
def test_ssh_step_risk_level_is_medium(self):
"""SSH 步驟預設 MEDIUM 風險"""
incident = _make_incident()
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh {host} 'docker restart minio'"
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert steps[0].risk_level == RiskLevel.MEDIUM
def test_ssh_step_number_is_1(self):
"""SSH 步驟編號從 1 開始"""
incident = _make_incident()
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh 192.168.0.188 'docker restart minio'"
svc = self._svc()
steps = svc._extract_repair_steps(incident)
assert steps[0].step_number == 1
# =============================================================================
# _generate_name — [SSH] 前綴
# =============================================================================
class TestGenerateNameSSH:
"""SSH 修復時名稱包含 [SSH] 前綴"""
def _svc(self) -> PlaybookService:
return PlaybookService.__new__(PlaybookService)
def test_ssh_name_has_prefix(self):
"""SSH 修復 → 名稱含 [SSH]"""
incident = _make_incident(alert_name="MinioDown", affected_services=["minio"])
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh 192.168.0.188 'docker restart minio'"
svc = self._svc()
name = svc._generate_name(incident)
assert name.startswith("[SSH]")
assert "MinioDown" in name
def test_kubectl_name_no_prefix(self):
"""kubectl 修復 → 名稱無 [SSH] 前綴"""
incident = _make_incident(alert_name="KubePodCrashLooping")
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = (
"kubectl rollout restart deployment/awoooi-api -n prod"
)
svc = self._svc()
name = svc._generate_name(incident)
assert not name.startswith("[SSH]")
def test_no_outcome_no_prefix(self):
"""無 outcome → 名稱無前綴"""
incident = _make_incident()
incident.outcome = None
svc = self._svc()
name = svc._generate_name(incident)
assert not name.startswith("[SSH]")
# =============================================================================
# _extract_tags — ssh / host_layer 標籤
# =============================================================================
class TestExtractTagsSSH:
"""SSH 修復時自動加 ssh/host_layer 標籤"""
def _svc(self) -> PlaybookService:
return PlaybookService.__new__(PlaybookService)
def test_ssh_tags_added(self):
"""SSH last_repair_action → tags 含 ssh + host_layer"""
incident = _make_incident(affected_services=["minio"])
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = "ssh 192.168.0.188 'docker restart minio'"
svc = self._svc()
tags = svc._extract_tags(incident)
assert "ssh" in tags
assert "host_layer" in tags
def test_non_ssh_no_ssh_tag(self):
"""kubectl 修復 → 無 ssh 標籤"""
incident = _make_incident()
incident.outcome = IncidentOutcome()
incident.outcome.learning_notes = (
"kubectl rollout restart deployment/awoooi-api -n prod"
)
svc = self._svc()
tags = svc._extract_tags(incident)
assert "ssh" not in tags
def test_no_outcome_no_ssh_tag(self):
"""無 outcome → 無 ssh 標籤"""
incident = _make_incident()
incident.outcome = None
svc = self._svc()
tags = svc._extract_tags(incident)
assert "ssh" not in tags