93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
def test_auto_heal_blocks_unsafe_ssh_commands():
|
|
from services.auto_heal_service import AutoHealService
|
|
|
|
svc = AutoHealService()
|
|
|
|
assert svc._is_allowed_ssh_argv(["docker", "ps"]) is True
|
|
assert svc._is_allowed_ssh_argv(["df", "-h"]) is True
|
|
assert svc._is_allowed_ssh_argv(["docker", "restart", "momo-pro-system"]) is False
|
|
assert svc._is_allowed_ssh_argv(["docker", "logs", "momo-db"]) is False
|
|
assert svc._is_allowed_ssh_argv(["bash", "-lc", "whoami"]) is False
|
|
|
|
|
|
def test_auto_heal_code_fix_writes_audit(monkeypatch):
|
|
from services.auto_heal_service import AutoHealResult, AutoHealService
|
|
|
|
svc = AutoHealService()
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
svc,
|
|
"_handle_code_fix",
|
|
lambda playbook, context: AutoHealResult(True, "CODE_FIX", "fixed"),
|
|
)
|
|
monkeypatch.setattr(
|
|
svc,
|
|
"_alert_and_store",
|
|
lambda playbook, context, result, duration_ms=0.0: calls.append((playbook, context, result, duration_ms)),
|
|
)
|
|
monkeypatch.setattr("services.auto_heal_service._find_playbook", lambda error_type: {
|
|
"id": 99,
|
|
"name": "Code fix",
|
|
"action_type": "CODE_FIX",
|
|
"action_params": {},
|
|
"max_retries": 1,
|
|
"cooldown_min": 0,
|
|
})
|
|
monkeypatch.setattr(svc, "_ensure_incident", lambda error_type, context: 123)
|
|
|
|
result = svc.handle_exception("code_exception", {"target_file": "services/example.py"})
|
|
|
|
assert result.success is True
|
|
assert calls
|
|
assert calls[0][2].action == "CODE_FIX"
|
|
|
|
|
|
def test_incident_model_keeps_legacy_and_current_columns():
|
|
from database.manager import Base
|
|
|
|
columns = set(Base.metadata.tables["incidents"].columns.keys())
|
|
|
|
assert {
|
|
"error_traceback",
|
|
"traceback_str",
|
|
"playbook_id",
|
|
"matched_playbook_id",
|
|
"resolved_at",
|
|
} <= columns
|
|
|
|
|
|
def test_auto_heal_status_update_backfills_dual_playbook_columns(monkeypatch):
|
|
from services.auto_heal_service import AutoHealResult, AutoHealService
|
|
|
|
captured = {}
|
|
|
|
class Session:
|
|
def execute(self, stmt, params):
|
|
captured["sql"] = str(stmt)
|
|
captured["params"] = params
|
|
|
|
def commit(self):
|
|
captured["committed"] = True
|
|
|
|
def rollback(self):
|
|
captured["rolled_back"] = True
|
|
|
|
def close(self):
|
|
captured["closed"] = True
|
|
|
|
monkeypatch.setattr("services.auto_heal_service.get_session", lambda: Session())
|
|
|
|
svc = AutoHealService()
|
|
svc._update_incident_status(
|
|
{"incident_id": 123, "matched_playbook_id": 77},
|
|
AutoHealResult(True, "ALERT_ONLY", "ok"),
|
|
)
|
|
|
|
assert "matched_playbook_id" in captured["sql"]
|
|
assert "playbook_id" in captured["sql"]
|
|
assert "resolved_at" in captured["sql"]
|
|
assert captured["params"]["status"] == "closed"
|
|
assert captured["params"]["playbook_id"] == 77
|
|
assert captured["params"]["incident_id"] == 123
|
|
assert captured["committed"] is True
|