fix(api): connect approval execution truth chain
This commit is contained in:
@@ -117,6 +117,56 @@ async def test_telegram_approval_schedules_executor_after_required_signature(mon
|
||||
assert op_log_repo.rows[0]["kwargs"]["action_detail"] == "approve"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_telegram_approval_suppresses_executor_for_no_action(monkeypatch):
|
||||
approval_id = "55555555-5555-5555-5555-555555555555"
|
||||
approval = SimpleNamespace(
|
||||
id=UUID(approval_id),
|
||||
status=SimpleNamespace(value="approved"),
|
||||
incident_id="INC-20260611-NOEXEC",
|
||||
action="NO_ACTION - REPAIR_CANDIDATE_MISSING: LLM 分析失敗",
|
||||
)
|
||||
finalizer_calls: list[dict] = []
|
||||
op_log_repo = _FakeAlertOperationLogRepository()
|
||||
|
||||
async def fake_finalize(*, approval, execution_triggered: bool) -> bool:
|
||||
finalizer_calls.append({
|
||||
"approval_id": str(approval.id),
|
||||
"execution_triggered": execution_triggered,
|
||||
})
|
||||
return False
|
||||
|
||||
fake_gateway = _FakeGateway({
|
||||
"success": True,
|
||||
"action": "approve",
|
||||
"approval_id": approval_id,
|
||||
"user": {"id": 42, "username": "ops"},
|
||||
})
|
||||
monkeypatch.setattr(telegram_api, "get_telegram_gateway", lambda: fake_gateway)
|
||||
monkeypatch.setattr(
|
||||
telegram_api,
|
||||
"get_approval_service",
|
||||
lambda: _FakeApprovalService(approval, execution_triggered=True),
|
||||
)
|
||||
monkeypatch.setattr(telegram_api, "_finalize_telegram_approval", fake_finalize)
|
||||
monkeypatch.setattr(
|
||||
"src.repositories.alert_operation_log_repository.get_alert_operation_log_repository",
|
||||
lambda: op_log_repo,
|
||||
)
|
||||
|
||||
result = await telegram_api.telegram_webhook(_callback_update(f"approve:{approval_id}:ts:nonce"))
|
||||
|
||||
assert result["ok"] is True
|
||||
assert result["message"] == "ApprovedWithoutExecution"
|
||||
assert result["execution_triggered"] is True
|
||||
assert result["execution_scheduled"] is False
|
||||
assert result["execution_suppressed"] is True
|
||||
assert finalizer_calls == [{
|
||||
"approval_id": approval_id,
|
||||
"execution_triggered": True,
|
||||
}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_telegram_approval_duplicate_does_not_schedule_executor(monkeypatch):
|
||||
approval_id = "33333333-3333-3333-3333-333333333333"
|
||||
@@ -242,3 +292,33 @@ async def test_finalize_telegram_approval_runs_executor_task(monkeypatch):
|
||||
assert scheduled is True
|
||||
await telegram_api.asyncio.sleep(0)
|
||||
assert executed == ["33333333-3333-3333-3333-333333333333"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_finalize_telegram_approval_does_not_schedule_no_action(monkeypatch):
|
||||
executed: list[str] = []
|
||||
approval = SimpleNamespace(
|
||||
id=UUID("66666666-6666-6666-6666-666666666666"),
|
||||
incident_id="INC-20260611-NOOP",
|
||||
action="OBSERVE",
|
||||
)
|
||||
|
||||
class _FakeExecutionService:
|
||||
async def execute_approved_action(self, received_approval):
|
||||
executed.append(str(received_approval.id))
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
telegram_api,
|
||||
"get_execution_service",
|
||||
lambda: _FakeExecutionService(),
|
||||
)
|
||||
|
||||
scheduled = await telegram_api._finalize_telegram_approval(
|
||||
approval=approval,
|
||||
execution_triggered=True,
|
||||
)
|
||||
|
||||
assert scheduled is False
|
||||
await telegram_api.asyncio.sleep(0)
|
||||
assert executed == []
|
||||
|
||||
Reference in New Issue
Block a user