fix(alerts): correct telegram execution truth
Some checks failed
CD Pipeline / tests (push) Failing after 52s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
Code Review / ai-code-review (push) Successful in 11s

This commit is contained in:
Your Name
2026-05-31 13:58:21 +08:00
parent 943a6feacf
commit e2ab879636
15 changed files with 624 additions and 49 deletions

View File

@@ -15,12 +15,18 @@ class _FakeGateway:
class _FakeApprovalService:
def __init__(self, approval, execution_triggered: bool) -> None:
def __init__(
self,
approval,
execution_triggered: bool,
sign_message: str = "Approval complete",
) -> None:
self.approval = approval
self.execution_triggered = execution_triggered
self.sign_message = sign_message
async def sign_approval(self, **_kwargs):
return self.approval, "Approval complete", self.execution_triggered
return self.approval, self.sign_message, self.execution_triggered
async def reject_approval(self, **_kwargs):
return self.approval, "Approval rejected"
@@ -100,6 +106,59 @@ 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_duplicate_does_not_schedule_executor(monkeypatch):
approval_id = "33333333-3333-3333-3333-333333333333"
approval = SimpleNamespace(
id=UUID(approval_id),
status=SimpleNamespace(value="execution_success"),
incident_id="INC-20260531-DUPE",
)
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 True
monkeypatch.setattr(
telegram_api,
"get_telegram_gateway",
lambda: _FakeGateway({
"success": True,
"action": "approve",
"approval_id": approval_id,
"user": {"id": 42, "username": "ops"},
}),
)
monkeypatch.setattr(
telegram_api,
"get_approval_service",
lambda: _FakeApprovalService(
approval,
execution_triggered=False,
sign_message="Cannot sign: status is execution_success",
),
)
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"] == "Already processed"
assert result["execution_triggered"] is False
assert result["execution_scheduled"] is False
assert finalizer_calls == []
assert op_log_repo.rows[0]["kwargs"]["action_detail"] == "approve_duplicate"
@pytest.mark.asyncio
async def test_telegram_rejection_syncs_incident_state(monkeypatch):
approval_id = "22222222-2222-2222-2222-222222222222"