wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(sre): enforce typed host ansible handoffs

Browse Source
This commit is contained in:
Your Name
2026-07-18 23:11:04 +08:00
parent 83fe799f50
commit 5b2547e4ec
24 changed files with 2255 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
apps/api/tests/test_ansible_verified_closure.py
View File

@@ -4,6 +4,7 @@ import inspect
import json
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
@@ -92,6 +93,46 @@ def _verified_result() -> service.AnsibleRunResult:
)
@pytest.mark.asyncio
async def test_terminal_apply_replaces_queued_approval_truth(
monkeypatch: pytest.MonkeyPatch,
) -> None:
approval = SimpleNamespace(
status=service.ApprovalStatus.APPROVED,
resolved_at=None,
rejection_reason=None,
extra_metadata={
"execution_kind": "host_ansible_check_mode_queued",
"execution_state": "queued",
"repair_executed": False,
"repair_verified": False,
"incident_closure_allowed": False,
},
)
db = _SequenceDB(_MappingResult(scalar=approval))
@asynccontextmanager
async def fake_get_db_context(_project_id):
yield db
monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
written = await service._finalize_controlled_approval_projection(
_claim(),
_verified_result(),
apply_op_id="00000000-0000-0000-0000-000000000104",
project_id="awoooi",
)
assert written is True
assert approval.status == service.ApprovalStatus.EXECUTION_SUCCESS
assert approval.extra_metadata["execution_kind"] == "controlled_apply"
assert approval.extra_metadata["execution_state"] == "completed"
assert approval.extra_metadata["repair_executed"] is True
assert approval.extra_metadata["repair_verified"] is True
assert approval.extra_metadata["incident_closure_allowed"] is True
def _controlled_result_payload() -> dict:
return {
"chat_id": "sre-chat",

63
apps/api/tests/test_approval_controlled_handoff_pending.py Normal file
View File

@@ -0,0 +1,63 @@
from contextlib import asynccontextmanager
from types import SimpleNamespace
from uuid import uuid4
import pytest
from src.models.approval import ApprovalStatus
from src.services import approval_db as approval_db_module
from src.services.approval_db import ApprovalDBService
@pytest.mark.asyncio
async def test_pending_handoff_preserves_approved_status_and_records_truth(
monkeypatch: pytest.MonkeyPatch,
) -> None:
record = SimpleNamespace(
status=ApprovalStatus.APPROVED,
extra_metadata={"existing": "kept"},
)
class FakeResult:
def scalar_one_or_none(self):
return record
class FakeDB:
async def execute(self, _query):
return FakeResult()
async def flush(self):
return None
@asynccontextmanager
async def fake_db_context():
yield FakeDB()
monkeypatch.setattr(
approval_db_module,
"get_db_context",
fake_db_context,
)
recorded = await ApprovalDBService().record_controlled_handoff_pending(
uuid4(),
execution_kind="host_ansible_check_mode_queued",
receipt={
"automation_run_id": "run-host-188",
"single_writer_executor": "awoooi-ansible-executor-broker",
},
)
assert recorded is True
assert record.status == ApprovalStatus.APPROVED
assert record.extra_metadata == {
"existing": "kept",
"execution_kind": "host_ansible_check_mode_queued",
"execution_state": "queued",
"repair_attempted": False,
"repair_executed": False,
"repair_verified": False,
"incident_closure_allowed": False,
"automation_run_id": "run-host-188",
"single_writer_executor": "awoooi-ansible-executor-broker",
}

248
apps/api/tests/test_approval_execution_mcp_audit.py
View File

@@ -1,7 +1,9 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock
from uuid import UUID
import pytest
@@ -10,6 +12,7 @@ from src.models.approval import ApprovalRequest, RiskLevel
from src.plugins.mcp.gateway import GateApprovalError
from src.plugins.mcp.interfaces import MCPToolResult
from src.services.approval_execution import ApprovalExecutionService
from src.services.executor import ExecutionResult, OperationType
class FakeRedis:
@@ -20,6 +23,251 @@ class FakeRedis:
self.set_calls.append((key, value, ex))
@pytest.mark.asyncio
async def test_systemctl_approval_queues_ansible_without_ssh_gateway(
monkeypatch: pytest.MonkeyPatch,
) -> None:
queue = AsyncMock(
return_value={
"status": "controlled_check_mode_queued",
"queued": True,
"automation_run_id": "run-systemd-approval",
"single_writer_executor": "awoooi-ansible-executor-broker",
"runtime_write_performed": False,
"repair_executed": False,
"repair_verified": False,
}
)
gateway = AsyncMock(
side_effect=AssertionError("systemctl must not use the SSH gateway")
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action_for_incident_id",
queue,
)
monkeypatch.setattr(
ApprovalExecutionService,
"_execute_ssh_tool_via_gateway",
gateway,
)
approval = ApprovalRequest(
action="ssh 192.168.0.110 'systemctl restart node_exporter'",
description="typed Host/systemd approval",
risk_level=RiskLevel.MEDIUM,
requested_by="test",
required_signatures=0,
incident_id="INC-NODE-EXPORTER-110",
)
result = await ApprovalExecutionService()._execute_ssh_host_action(
approval=approval,
host="192.168.0.110",
)
assert result.success is True
assert result.k8s_response is not None
assert result.k8s_response["controlled_execution_queued"] is True
assert result.k8s_response["runtime_write_performed"] is False
assert result.k8s_response["repair_executed"] is False
queue.assert_awaited_once()
gateway.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"action",
[
"ssh://root@192.168.0.110/systemctl%20start%20node_exporter",
"ssh 192.168.0.110 'systemctl restart node_exporter || true'",
(
"ssh 192.168.0.110 'docker restart exporter "
"|| systemctl restart node_exporter'"
),
],
)
async def test_unbounded_systemd_approval_never_reaches_any_executor(
monkeypatch: pytest.MonkeyPatch,
action: str,
) -> None:
queue = AsyncMock(
side_effect=AssertionError("invalid systemd action must not be queued")
)
gateway = AsyncMock(
side_effect=AssertionError("invalid systemd action must not use SSH MCP")
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action_for_incident_id",
queue,
)
monkeypatch.setattr(
ApprovalExecutionService,
"_execute_ssh_tool_via_gateway",
gateway,
)
approval = ApprovalRequest(
action=action,
description="unbounded Host/systemd approval",
risk_level=RiskLevel.MEDIUM,
requested_by="test",
required_signatures=0,
incident_id="INC-NODE-EXPORTER-110",
)
result = await ApprovalExecutionService()._execute_ssh_host_action(
approval=approval,
host="192.168.0.110",
)
assert result.success is False
assert "unbounded_or_unallowlisted" in str(result.error)
queue.assert_not_awaited()
gateway.assert_not_awaited()
@pytest.mark.asyncio
async def test_queued_host_approval_does_not_close_or_learn_as_repair(
monkeypatch: pytest.MonkeyPatch,
) -> None:
approval = ApprovalRequest(
action="ssh 192.168.0.110 'systemctl restart node_exporter'",
description="typed Host/systemd approval",
risk_level=RiskLevel.MEDIUM,
requested_by="operator",
required_signatures=0,
incident_id="INC-NODE-EXPORTER-110",
)
handoff = {
"status": "controlled_check_mode_queued",
"queued": True,
"automation_run_id": "run-systemd-approval",
"single_writer_executor": "awoooi-ansible-executor-broker",
"runtime_write_performed": False,
"repair_executed": False,
"repair_verified": False,
}
execute_ssh = AsyncMock(
return_value=ExecutionResult(
success=True,
message="host_ansible_controlled_executor queued",
operation_type=OperationType.SSH_HOST,
target_resource="192.168.0.110",
namespace="host",
duration_ms=1,
k8s_response={
"controlled_execution_queued": True,
"runtime_write_performed": False,
"controlled_handoff": handoff,
},
)
)
update_status = AsyncMock()
record_pending = AsyncMock(return_value=True)
timeline = SimpleNamespace(add_event=AsyncMock())
log_completed = AsyncMock()
alert_completed = AsyncMock()
push_result = AsyncMock()
monkeypatch.setattr(
"src.services.approval_execution.get_approval_service",
lambda: SimpleNamespace(
update_execution_status=update_status,
record_controlled_handoff_pending=record_pending,
),
)
monkeypatch.setattr(
"src.services.approval_execution.get_timeline_service",
lambda: timeline,
)
monkeypatch.setattr(
"src.services.approval_execution.get_executor",
lambda: object(),
)
monkeypatch.setattr(
"src.services.resource_resolver.get_resource_resolver",
lambda: SimpleNamespace(
resolve=AsyncMock(
return_value=SimpleNamespace(
success=False,
resource_name=None,
candidates=[],
)
)
),
)
monkeypatch.setattr(
ApprovalExecutionService,
"_execute_ssh_host_action",
execute_ssh,
)
monkeypatch.setattr(
ApprovalExecutionService,
"_log_aol_started",
AsyncMock(return_value="aol-op-id"),
)
monkeypatch.setattr(
ApprovalExecutionService,
"_log_alert_execution_started",
AsyncMock(),
)
monkeypatch.setattr(
ApprovalExecutionService,
"_log_aol_completed",
log_completed,
)
monkeypatch.setattr(
ApprovalExecutionService,
"_log_alert_execution_completed",
alert_completed,
)
monkeypatch.setattr(
ApprovalExecutionService,
"_push_execution_result_to_alert",
push_result,
)
learning = AsyncMock(
side_effect=AssertionError("queued handoff must not trigger learning")
)
resolve = AsyncMock(
side_effect=AssertionError("queued handoff must not resolve incident")
)
monkeypatch.setattr(
ApprovalExecutionService,
"_trigger_learning",
learning,
)
monkeypatch.setattr(
"src.services.incident_service.get_incident_service",
lambda: SimpleNamespace(resolve_incident=resolve),
)
result = await ApprovalExecutionService().execute_approved_action(approval)
assert result is False
update_status.assert_not_awaited()
record_pending.assert_awaited_once_with(
approval.id,
execution_kind="host_ansible_check_mode_queued",
receipt=handoff,
)
assert timeline.add_event.await_args.kwargs["status"] == "pending"
assert "尚未修復" in timeline.add_event.await_args.kwargs["title"]
alert_completed.assert_not_awaited()
assert log_completed.await_args.kwargs["status"] == "dry_run"
assert log_completed.await_args.kwargs["output"]["incident_closure_allowed"] is False
push_result.assert_awaited_once_with(
approval,
success=False,
error=None,
execution_kind="host_ansible_check_mode_queued",
repair_executed=False,
repair_attempted=False,
)
learning.assert_not_awaited()
resolve.assert_not_awaited()
@pytest.mark.asyncio
async def test_ssh_approval_execution_projects_approval_into_gateway(
monkeypatch: pytest.MonkeyPatch,

66
apps/api/tests/test_auto_repair_api.py Normal file
View File

@@ -0,0 +1,66 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from src.api.v1 import auto_repair as auto_repair_api
@pytest.mark.asyncio
async def test_execute_api_preserves_controlled_queue_truth(
monkeypatch: pytest.MonkeyPatch,
) -> None:
incident = object()
playbook = object()
incident_service = SimpleNamespace(
get_from_working_memory=AsyncMock(return_value=incident)
)
playbook_service = SimpleNamespace(
get_by_id=AsyncMock(return_value=playbook)
)
repair_service = SimpleNamespace(
execute_auto_repair=AsyncMock(
return_value=SimpleNamespace(
success=False,
status="queued",
repair_executed=False,
repair_verified=False,
incident_id="INC-HOST-QUEUED",
playbook_id="PB-HOST-ANSIBLE",
executed_steps=[
"Step 1: systemctl restart node_exporter -> QUEUED"
],
error=None,
execution_time_ms=7,
)
)
)
monkeypatch.setattr(
auto_repair_api,
"get_incident_service",
lambda: incident_service,
)
monkeypatch.setattr(
"src.services.playbook_service.get_playbook_service",
lambda: playbook_service,
)
monkeypatch.setattr(
auto_repair_api,
"get_auto_repair_service",
lambda: repair_service,
)
response = await auto_repair_api.execute_auto_repair(
auto_repair_api.ExecuteRequest(
incident_id="INC-HOST-QUEUED",
playbook_id="PB-HOST-ANSIBLE",
force=True,
),
None, # type: ignore[arg-type]
)
assert response.success is False
assert response.status == "queued"
assert response.repair_executed is False
assert response.repair_verified is False
assert response.error is None

358
apps/api/tests/test_auto_repair_service.py
View File

@@ -8,6 +8,8 @@ Auto Repair Service Tests - #8 自動升級決策
建立者: Claude Code (#8 自動升級決策)
"""
from unittest.mock import AsyncMock
import pytest
from src.models.incident import Incident, IncidentStatus, Severity, Signal
@@ -179,6 +181,362 @@ class TestAutoRepairService:
assert decision.risk_level == RiskLevel.HIGH
assert decision.blocked_by is None
@pytest.mark.asyncio
async def test_systemctl_step_queues_host_ansible_without_legacy_agent(
self,
service,
monkeypatch,
):
incident = create_test_incident(
severity=Severity.P2,
alert_category="infrastructure",
alert_name="NodeExporterDown",
)
incident.affected_services = ["node-exporter-110"]
incident.signals[0].labels.update(
{
"alertname": "NodeExporterDown",
"host": "110",
"instance": "192.168.0.110",
"service": "node-exporter-110",
}
)
queue = AsyncMock(
return_value={
"status": "controlled_check_mode_queued",
"queued": True,
"automation_run_id": "run-node-exporter-110",
"runtime_write_performed": False,
"repair_executed": False,
"repair_verified": False,
}
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action",
queue,
)
step = RepairStep(
step_number=1,
action_type=ActionType.SSH_COMMAND,
command=(
"ssh {host} 'sudo systemctl restart node_exporter.service'"
),
risk_level=RiskLevel.MEDIUM,
requires_approval=False,
)
result = await service._execute_step(incident, step)
assert result == "QUEUED: host_ansible_executor:run-node-exporter-110"
queue.assert_awaited_once()
assert queue.await_args.kwargs["requested_host"] == "110"
@pytest.mark.asyncio
async def test_systemctl_uri_queues_with_explicit_host(
self,
service,
monkeypatch,
):
incident = create_test_incident(
alert_category="infrastructure",
alert_name="NodeExporterDown",
)
queue = AsyncMock(
return_value={
"status": "controlled_check_mode_queued",
"queued": True,
"automation_run_id": "run-node-exporter-188",
}
)
legacy = AsyncMock(
side_effect=AssertionError("systemd URI must not use direct SSH")
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action",
queue,
)
monkeypatch.setattr(
"src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
legacy,
)
step = RepairStep(
step_number=1,
action_type=ActionType.SSH_COMMAND,
command=(
"ssh://ollama@192.168.0.188/"
"systemctl%20restart%20node_exporter.service"
),
risk_level=RiskLevel.MEDIUM,
)
result = await service._execute_step(incident, step)
assert result == "QUEUED: host_ansible_executor:run-node-exporter-188"
assert queue.await_args.kwargs["requested_host"] == "192.168.0.188"
legacy.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"command,reason",
[
(
"ssh://root@192.168.0.188/systemctl%20start%20node_exporter",
"systemd_action_not_allowlisted:start",
),
(
"ssh://root@192.168.0.188/systemctl%20stop%20node_exporter",
"systemd_action_not_allowlisted:stop",
),
(
"ssh://root@192.168.0.188/systemctl%20reload%20node_exporter",
"systemd_action_not_allowlisted:reload",
),
(
"ssh://root@192.168.0.188/systemctl%20restart%20--no-block",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'sudo -n systemctl restart node_exporter'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl restart node_exporter || true'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl restart node_exporter; true'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl restart $(hostname)'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl restart\nnode_exporter'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl try-restart node_exporter'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh 192.168.0.188 'systemctl enable node_exporter'",
"unbounded_systemd_mutation_rejected",
),
(
"ssh://root@192.168.0.188/"
"sys%22%22temctl%20restart%20node_exporter",
"unbounded_systemd_mutation_rejected",
),
(
"ssh://root@192.168.0.188/"
"syst%5Cemctl%20restart%20node_exporter",
"unbounded_systemd_mutation_rejected",
),
(
"ssh://root@192.168.0.188/"
"service%20node_exporter%20restart",
"unbounded_systemd_mutation_rejected",
),
],
)
async def test_unallowlisted_systemd_mutations_never_reach_direct_ssh(
self,
service,
monkeypatch,
command,
reason,
):
incident = create_test_incident(
alert_category="infrastructure",
alert_name="NodeExporterDown",
)
queue = AsyncMock()
legacy = AsyncMock(
side_effect=AssertionError("systemd mutation must fail closed")
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action",
queue,
)
monkeypatch.setattr(
"src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
legacy,
)
step = RepairStep(
step_number=1,
action_type=ActionType.SSH_COMMAND,
command=command,
risk_level=RiskLevel.MEDIUM,
)
result = await service._execute_step(incident, step)
assert result == f"FAILED: host_ansible_controlled_executor:{reason}"
queue.assert_not_awaited()
legacy.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"command",
[
"ssh://wooo@192.168.0.110/echo%20ready",
"ansible://192.168.0.188/untyped_legacy.yml",
"openclaw://docker-110/sentry",
"unsupported://192.168.0.110/restart",
],
)
async def test_untyped_host_executor_fallback_is_disabled(
self,
service,
monkeypatch,
command,
):
incident = create_test_incident(
alert_category="infrastructure",
alert_name="HostRepairRequested",
)
legacy = AsyncMock(
side_effect=AssertionError("untyped commands must not use direct host executor")
)
monkeypatch.setattr(
"src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
legacy,
)
step = RepairStep(
step_number=1,
action_type=ActionType.SSH_COMMAND,
command=command,
risk_level=RiskLevel.MEDIUM,
)
result = await service._execute_step(incident, step)
assert result == (
"FAILED: legacy_direct_host_executor_disabled_use_typed_executor"
)
legacy.assert_not_awaited()
@pytest.mark.asyncio
async def test_controlled_queue_is_not_recorded_as_repair_success(
self,
service,
mock_playbook_service,
monkeypatch,
):
incident = create_test_incident(
severity=Severity.P2,
alert_category="infrastructure",
alert_name="NodeExporterDown",
)
playbook = Playbook(
playbook_id="PB-HOST-ANSIBLE-QUEUED",
name="Host Ansible controlled repair",
description="queue exact Host/systemd repair",
status=PlaybookStatus.APPROVED,
symptom_pattern=SymptomPattern(
alert_names=["NodeExporterDown"],
affected_services=["node-exporter-110"],
),
repair_steps=[
RepairStep(
step_number=1,
action_type=ActionType.SSH_COMMAND,
command="ssh {host} 'systemctl restart node_exporter'",
risk_level=RiskLevel.MEDIUM,
)
],
)
mock_playbook_service.add_playbook(playbook)
monkeypatch.setattr(
service,
"_execute_step",
AsyncMock(
return_value="QUEUED: host_ansible_executor:run-pending"
),
)
repair_counter = AsyncMock()
monkeypatch.setattr(
"src.services.auto_repair_service.record_global_repair_action",
repair_counter,
)
result = await service.execute_auto_repair(incident, playbook)
assert result.success is False
assert result.status == "queued"
assert result.repair_executed is False
assert result.repair_verified is False
assert playbook.success_count == 0
assert playbook.failure_count == 0
repair_counter.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"host_command",
[
"systemctl restart node_exporter",
"ssh {host} 'sys$@temctl restart node_exporter'",
"ssh {host} 'sys$*temctl restart node_exporter'",
],
)
async def test_host_playbook_cannot_execute_another_domain_first(
self,
service,
mock_playbook_service,
monkeypatch,
host_command,
):
incident = create_test_incident(
alert_category="infrastructure",
alert_name="NodeExporterDown",
)
playbook = Playbook(
playbook_id="PB-MIXED-DOMAIN-BLOCKED",
name="mixed domain playbook",
description="must fail before any runtime action",
status=PlaybookStatus.APPROVED,
symptom_pattern=SymptomPattern(
alert_names=["NodeExporterDown"],
affected_services=["node-exporter-110"],
),
repair_steps=[
RepairStep(
step_number=1,
action_type=ActionType.KUBECTL,
command="kubectl rollout restart deployment/api",
risk_level=RiskLevel.MEDIUM,
),
RepairStep(
step_number=2,
action_type=ActionType.SSH_COMMAND,
command=host_command,
risk_level=RiskLevel.MEDIUM,
),
],
)
mock_playbook_service.add_playbook(playbook)
execute_step = AsyncMock(
side_effect=AssertionError("mixed domain must fail before execution")
)
repair_counter = AsyncMock()
monkeypatch.setattr(service, "_execute_step", execute_step)
monkeypatch.setattr(
"src.services.auto_repair_service.record_global_repair_action",
repair_counter,
)
result = await service.execute_auto_repair(incident, playbook)
assert result.success is False
assert result.status == "failed"
assert "cross_domain_or_multi_step" in str(result.error)
execute_step.assert_not_awaited()
repair_counter.assert_not_awaited()
@pytest.mark.asyncio
async def test_evaluate_allows_p0_severity_with_approved_playbook(
self,

89
apps/api/tests/test_awooop_truth_chain_service.py
View File

@@ -859,6 +859,95 @@ def test_reconciliation_ignores_no_action_audit_rows_as_execution() -> None:
assert "incident_open_after_successful_execution" not in codes
def test_reconciliation_keeps_host_ansible_queue_pending_not_executed() -> None:
reconciliation = build_incident_reconciliation(
incident={"incident_id": "INC-HOST-QUEUED", "status": "INVESTIGATING"},
approvals=[
{
"id": "approval-host-queued",
"status": "APPROVED",
"action": "ssh 192.168.0.188 'systemctl restart node_exporter'",
"resolved_at": "2026-07-18T01:00:00+00:00",
"extra_metadata": {
"execution_kind": "host_ansible_check_mode_queued",
"execution_state": "queued",
"repair_attempted": False,
"repair_executed": False,
"repair_verified": False,
"automation_run_id": "run-host-queued",
},
}
],
evidence_rows=[{"sensors_attempted": 3, "sensors_succeeded": 3}],
automation_ops=[
{
"operation_type": "playbook_executed",
"status": "dry_run",
"actor": "approval_execution",
"output_execution_kind": "host_ansible_check_mode_queued",
"output_repair_executed": False,
},
{
"operation_type": "ansible_candidate_matched",
"status": "dry_run",
"actor": "awoooi-ansible-executor-broker",
},
],
auto_repair_executions=[],
timeline_events=[{"event_type": "exec", "status": "pending"}],
)
codes = {row["code"] for row in reconciliation["mismatches"]}
assert reconciliation["facts"]["effective_execution_records"] == 0
assert reconciliation["facts"]["controlled_handoff_pending"] is True
assert "incident_open_after_successful_execution" not in codes
assert "incident_open_after_approval_resolved" not in codes
assert "approval_approved_without_execution_record" not in codes
def test_reconciliation_counts_terminal_host_ansible_apply_after_queue() -> None:
reconciliation = build_incident_reconciliation(
incident={"incident_id": "INC-HOST-APPLIED", "status": "INVESTIGATING"},
approvals=[
{
"id": "approval-host-applied",
"status": "EXECUTION_SUCCESS",
"action": "systemctl restart node_exporter",
"resolved_at": "2026-07-18T01:05:00+00:00",
"extra_metadata": {
"execution_kind": "controlled_apply",
"execution_state": "queued",
"repair_attempted": True,
"repair_executed": True,
"post_verifier_passed": True,
"automation_run_id": "run-host-applied",
},
}
],
evidence_rows=[
{
"sensors_attempted": 3,
"sensors_succeeded": 3,
"verification_result": "success",
}
],
automation_ops=[
{
"operation_type": "ansible_apply_executed",
"status": "success",
"actor": "ansible_controlled_apply_worker",
}
],
auto_repair_executions=[],
timeline_events=[{"event_type": "exec", "status": "success"}],
)
codes = {row["code"] for row in reconciliation["mismatches"]}
assert reconciliation["facts"]["effective_execution_records"] == 1
assert reconciliation["facts"]["controlled_handoff_pending"] is False
assert "incident_open_after_successful_execution" in codes
def test_automation_quality_ignores_no_action_audit_rows_as_execution() -> None:
quality = build_automation_quality(
incident={"incident_id": "INC-1", "status": "RESOLVED"},

75
apps/api/tests/test_callback_dispatcher.py
View File

@@ -13,21 +13,22 @@ Phase 5 Sprint 5.0-5.1 Callback Dispatcher 單元測試
🔴 遵循「禁止 Mock 測試鐵律」: 用真實 spec registry，不 mock。
"""
from unittest.mock import AsyncMock
import pytest
from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
from src.services import callback_dispatcher as callback_dispatcher_module
from src.services.callback_dispatcher import (
_lookup_context,
_resolve_provider_name,
_resolve_template,
dispatch_action,
get_action_spec,
list_actions_for_category,
load_action_registry,
_lookup_context,
_resolve_provider_name,
_resolve_template,
)
# =============================================================================
# Registry loading
# =============================================================================
@@ -49,6 +50,25 @@ class TestRegistryLoading:
assert spec.mcp_provider, f"{name} missing mcp_provider"
assert spec.mcp_tool, f"{name} missing mcp_tool"
@pytest.mark.parametrize(
"action_name,service",
[
("host_restart_service", "{labels.service}"),
("reload_nginx", "nginx"),
],
)
def test_systemd_callbacks_use_only_controlled_host_ansible_route(
self,
action_name,
service,
):
spec = get_action_spec(action_name)
assert spec is not None
assert spec.mcp_provider == "controlled"
assert spec.mcp_tool == "host_ansible_reconcile"
assert spec.mcp_params["service"] == service
def test_secops_requires_multi_sig(self):
for sa in ("secops_isolate", "secops_block_ip", "secops_evict"):
spec = get_action_spec(sa)
@@ -356,3 +376,50 @@ async def test_dispatch_action_injects_mcp_audit_context(monkeypatch):
assert audit_context["flywheel_node"] == "operate"
assert audit_context["agent_role"] == "telegram_callback_dispatcher"
assert audit_context["operator_user_id"] == 12345
@pytest.mark.asyncio
async def test_host_restart_button_queues_ansible_without_ssh_provider(
monkeypatch,
):
queue = AsyncMock(
return_value={
"status": "controlled_check_mode_queued",
"queued": True,
"automation_run_id": "run-host-111",
"runtime_write_performed": False,
"repair_executed": False,
"repair_verified": False,
}
)
provider_lookup = AsyncMock(
side_effect=AssertionError("Host/systemd callback must not load SSH MCP")
)
monkeypatch.setattr(
"src.services.host_ansible_controlled_executor."
"queue_host_ansible_controlled_action_for_incident_id",
queue,
)
monkeypatch.setattr("src.plugins.mcp.registry.get_provider", provider_lookup)
result = await dispatch_action(
action_name="host_restart_service",
incident_id="INC-HOST-111",
user_id=12345,
labels={
"instance": "192.168.0.111",
"service": "ollama-local",
},
)
assert result.success is True
assert "已排入 Ansible check-mode" in result.result_text
assert "尚未執行修復" in result.result_text
queue.assert_awaited_once_with(
incident_id="INC-HOST-111",
requested_action="reconcile host service ollama-local",
risk_level="high",
source="telegram_callback_host_systemd",
requested_host="192.168.0.111",
)
provider_lookup.assert_not_awaited()

184
apps/api/tests/test_host_ansible_controlled_executor.py Normal file
View File

@@ -0,0 +1,184 @@
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from src.services.host_ansible_controlled_executor import (
queue_host_ansible_controlled_action,
)
def _ollama111_incident() -> SimpleNamespace:
return SimpleNamespace(
incident_id="INC-OLLAMA111-001",
project_id="awoooi",
alertname="Ollama111Unavailable",
alert_category="ai_provider",
severity=SimpleNamespace(value="P2"),
affected_services=["ollama-local"],
signals=[
SimpleNamespace(
alert_name="Ollama111Unavailable",
labels={
"alertname": "Ollama111Unavailable",
"host": "111",
"instance": "192.168.0.111",
"service": "ollama-local",
},
annotations={},
)
],
)
@pytest.mark.asyncio
async def test_host_action_queues_exact_typed_ansible_candidate() -> None:
enqueuer = AsyncMock(
return_value={
"schema_version": "ai_decision_controlled_executor_handoff_v1",
"status": "controlled_check_mode_queued",
"queued": True,
"side_effect_performed": False,
"single_writer_executor": "awoooi-ansible-executor-broker",
"automation_run_id": "run-ollama111",
"trace_id": "trace-ollama111",
"work_item_id": "AIA-SRE-008",
"active_blockers": [],
}
)
receipt = await queue_host_ansible_controlled_action(
incident=_ollama111_incident(),
requested_action="reconcile host service ollama-local",
risk_level="low",
source="test_host_systemd",
requested_host="host_111",
enqueuer=enqueuer,
)
assert receipt["queued"] is True
assert receipt["runtime_write_performed"] is False
assert receipt["repair_executed"] is False
assert receipt["repair_verified"] is False
assert receipt["incident_closure_allowed"] is False
assert receipt["typed_route"]["target_kind"] == "host_launchagent"
assert receipt["typed_route"]["executor"] == "host_ansible_executor"
assert receipt["typed_route"]["verifier"] == (
"ollama111_k3s_path_independent_verifier"
)
assert receipt["typed_route"]["allowed_catalog_ids"] == [
"ansible:111-ollama-fallback"
]
enqueuer.assert_awaited_once()
proposal = enqueuer.await_args.kwargs["proposal_data"]
assert proposal["risk_level"] == "medium"
assert proposal["cross_domain_fallback_allowed"] is False
assert proposal["check_mode_required_before_apply"] is True
@pytest.mark.asyncio
async def test_host_action_blocks_requested_host_mismatch_without_queue() -> None:
enqueuer = AsyncMock()
receipt = await queue_host_ansible_controlled_action(
incident=_ollama111_incident(),
requested_action="reconcile host service ollama-local",
risk_level="medium",
source="test_host_systemd",
requested_host="192.168.0.110",
enqueuer=enqueuer,
)
assert receipt["queued"] is False
assert receipt["status"] == "requested_host_typed_route_mismatch"
assert receipt["runtime_write_performed"] is False
enqueuer.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"malformed_receipt",
[
{"queued": True},
{
"schema_version": "ai_decision_controlled_executor_handoff_v1",
"status": "controlled_executor_queue_failed",
"queued": True,
"side_effect_performed": False,
"single_writer_executor": "awoooi-ansible-executor-broker",
"automation_run_id": "run-1",
"trace_id": "trace-1",
"work_item_id": "AIA-SRE-008",
"active_blockers": [],
},
{
"schema_version": "ai_decision_controlled_executor_handoff_v1",
"status": "controlled_check_mode_queued",
"queued": True,
"side_effect_performed": True,
"single_writer_executor": "awoooi-ansible-executor-broker",
"automation_run_id": "run-1",
"trace_id": "trace-1",
"work_item_id": "AIA-SRE-008",
"active_blockers": [],
},
],
)
async def test_malformed_broker_receipt_fails_closed(
malformed_receipt: dict,
) -> None:
receipt = await queue_host_ansible_controlled_action(
incident=_ollama111_incident(),
requested_action="reconcile host service ollama-local",
risk_level="medium",
source="test_host_systemd",
requested_host="192.168.0.111",
enqueuer=AsyncMock(return_value=malformed_receipt),
)
assert receipt["queued"] is False
assert receipt["status"] == "controlled_executor_receipt_invalid"
assert receipt["runtime_write_performed"] is False
assert receipt["incident_closure_allowed"] is False
@pytest.mark.asyncio
async def test_unknown_host_asset_fails_closed_without_fallback() -> None:
enqueuer = AsyncMock()
incident = SimpleNamespace(
incident_id="INC-UNKNOWN-HOST",
project_id="awoooi",
alertname="HostServiceDown",
alert_category="host_resource",
severity=SimpleNamespace(value="P2"),
affected_services=["mystery-daemon"],
signals=[
SimpleNamespace(
alert_name="HostServiceDown",
labels={
"alertname": "HostServiceDown",
"instance": "192.168.0.199",
"service": "mystery-daemon",
},
annotations={},
)
],
)
receipt = await queue_host_ansible_controlled_action(
incident=incident,
requested_action="reconcile host service mystery-daemon",
risk_level="medium",
source="test_host_systemd",
requested_host="192.168.0.199",
enqueuer=enqueuer,
)
assert receipt["queued"] is False
assert receipt["status"] == "asset_identity_unresolved"
assert receipt["typed_route"]["drift_work_item_id"].startswith(
"AIA-ASSET-DRIFT-"
)
enqueuer.assert_not_awaited()

44
apps/api/tests/test_ssh_provider_tools.py
View File

@@ -94,6 +94,50 @@ async def test_ssh_execute_normalizes_host_before_allowed_check(monkeypatch):
assert isinstance(captured["timeout"], int)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tool_name,parameters",
[
(
"ssh_systemctl_restart",
{"service": "node_exporter"},
),
("ssh_reload_nginx", {}),
],
)
async def test_systemd_tools_are_compatibility_only_no_ssh(
monkeypatch,
tool_name,
parameters,
):
provider = SSHProvider()
ssh_exec_called = False
async def fail_if_called(*_args, **_kwargs):
nonlocal ssh_exec_called
ssh_exec_called = True
raise AssertionError("direct systemctl SSH must remain disabled")
monkeypatch.setattr(provider, "_ssh_exec", fail_if_called)
result = await provider.execute(
tool_name,
{
"host": "192.168.0.110",
"trust_score": 1.0,
**parameters,
},
)
assert result.success is False
assert result.error == (
f"{tool_name}_disabled_use_host_ansible_controlled_executor"
)
assert ssh_exec_called is False
with pytest.raises(ValueError, match="host_ansible_controlled_executor"):
provider._build_command(tool_name, parameters)
def test_ssh_container_status_accepts_legacy_container_name_alias():
provider = SSHProvider()