Files
awoooi/apps/api/tests/test_backup_restore_alertmanager_ingress.py
ogt 6cf8429d17
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m32s
CD Pipeline / build-and-deploy (push) Successful in 8m2s
CD Pipeline / post-deploy-checks (push) Successful in 2m22s
fix(automation): enforce durable alert closure
Restore D037 durable incident readback without weakening database failure semantics.

Fence Agent99 dispatch and terminal learning to canonical identities, durable leases, verifier receipts, and reconciler-owned checkpoints.

Drain backup and restore legacy receipts in bounded cohorts with strict transport, claim, projection, and no-false-green controls.

Keep SSH service refusal visible while separating it from broker network-policy reachability.
2026-07-14 09:40:53 +08:00

398 lines
14 KiB
Python

from __future__ import annotations
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import UUID
import pytest
from fastapi import BackgroundTasks
from starlette.requests import Request
from src.api.v1 import webhooks
from src.services import backup_restore_alertmanager_ingress as ingress
from src.services.agent99_controlled_dispatch_ledger import (
build_agent99_dispatch_receipt_envelope,
)
from src.services.agent99_sre_bridge import bridge_alertmanager_to_agent99
def _request() -> Request:
return Request(
{
"type": "http",
"method": "POST",
"path": "/api/v1/webhooks/alertmanager",
"scheme": "http",
"server": ("testserver", 80),
"client": ("127.0.0.1", 50000),
"query_string": b"",
"headers": [],
}
)
def _signal_kwargs() -> dict[str, object]:
return {
"project_id": "awoooi",
"alert_id": "backup-alert-1",
"alertname": "BackupCredentialEscrowEvidenceMissing",
"severity": "critical",
"namespace": "awoooi-prod",
"target_resource": "backup_restore",
"message": "backup freshness and escrow evidence missing",
"labels": {
"event_type": "backup_restore_escrow_signal",
"lane": "backup_restore_escrow_triage",
},
"annotations": {"summary": "readback verifier required"},
"source_fingerprint": "source-fingerprint",
"source_event_fingerprint": "alertmanager-native-fingerprint",
"source_started_at": "2026-07-11T10:00:00Z",
"alert_category": "backup",
"notification_type": "TYPE-1",
"source_url": "http://alertmanager/graph",
}
def test_backup_ingress_owner_is_stable_per_source_occurrence() -> None:
first = ingress.build_backup_restore_ingress_owner(
project_id="awoooi",
source_fingerprint="source-fingerprint",
source_event_fingerprint="native-fingerprint",
source_started_at="2026-07-11T10:00:00Z",
)
replay = ingress.build_backup_restore_ingress_owner(
project_id="awoooi",
source_fingerprint="source-fingerprint",
source_event_fingerprint="native-fingerprint",
source_started_at="2026-07-11T10:00:00Z",
)
new_firing = ingress.build_backup_restore_ingress_owner(
project_id="awoooi",
source_fingerprint="source-fingerprint",
source_event_fingerprint="native-fingerprint",
source_started_at="2026-07-12T10:00:00Z",
)
assert first == replay
assert new_firing["approval_id"] != first["approval_id"]
assert first["route_id"] == "agent99:backup_health:BackupCheck"
assert first["single_dispatch_owner"] == "alertmanager_raw_signal"
assert first["telegram_role"] == "receipt_projection_only"
assert first["controlled_apply"] is False
assert first["read_only"] is True
@pytest.mark.asyncio
async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortcut(
monkeypatch: pytest.MonkeyPatch,
) -> None:
grouped = SimpleNamespace(is_grouped=False)
grouping = SimpleNamespace(evaluate=AsyncMock(return_value=grouped))
process = AsyncMock(return_value={"status": "queued"})
record = AsyncMock()
append = AsyncMock()
monkeypatch.setattr(webhooks, "get_alert_grouping_service", lambda: grouping)
monkeypatch.setattr(
"src.repositories.alert_operation_log_repository.get_alert_operation_log_repository",
lambda: SimpleNamespace(append=append),
)
monkeypatch.setattr(webhooks, "record_alertmanager_event", record)
monkeypatch.setattr(
webhooks,
"process_backup_restore_alertmanager_signal",
process,
)
monkeypatch.setattr(
webhooks,
"get_approval_service",
lambda: (_ for _ in ()).throw(
AssertionError("generic approval/TYPE-1 branch must not run")
),
)
tasks = BackgroundTasks()
payload = webhooks.AlertmanagerPayload(
status="firing",
alerts=[
webhooks.AlertmanagerAlert(
status="firing",
labels={
"alertname": "HostBackupFailed",
"severity": "warning",
"component": "backup_restore",
"namespace": "awoooi-prod",
},
annotations={
"summary": "backup freshness missing; readback required"
},
startsAt=datetime.now(UTC).isoformat(),
fingerprint="native-backup-fingerprint",
generatorURL="http://alertmanager/graph",
)
],
)
response = await webhooks.alertmanager_webhook(_request(), payload, tasks)
assert response.success is True
assert "durable read-only BackupCheck" in response.message
assert response.approval_created is False
assert process.await_count == 0
await tasks()
process.assert_awaited_once()
call = process.await_args.kwargs
assert call["alertname"] == "HostBackupFailed"
assert call["source_event_fingerprint"] == "native-backup-fingerprint"
assert call["notification_type"] == "TYPE-1"
assert call["alert_category"] == "backup"
assert call["target_resource"] == "backup_restore"
class _FakeApprovalService:
def __init__(self, calls: list[str]) -> None:
self.calls = calls
self.approval: SimpleNamespace | None = None
self.created_request = None
async def get_approval(self, approval_id: UUID):
self.calls.append("owner_read")
if self.approval is not None:
assert self.approval.id == approval_id
return self.approval
async def find_by_fingerprint(self, **_kwargs):
self.calls.append("legacy_owner_read")
return None
async def create_approval_with_fingerprint(self, *, request, fingerprint):
self.calls.append("owner_create")
self.created_request = request
assert fingerprint == "source-fingerprint"
self.approval = SimpleNamespace(
id=UUID(request.metadata["preallocated_approval_id"]),
incident_id=request.incident_id,
hit_count=1,
risk_level=request.risk_level,
)
return self.approval
async def increment_hit_count(self, approval_id: UUID):
self.calls.append("owner_recurrence")
assert self.approval is not None
assert self.approval.id == approval_id
self.approval.hit_count += 1
return self.approval
async def update_incident_id(self, approval_id: UUID, incident_id: str):
self.calls.append("incident_bind")
assert self.approval is not None
assert self.approval.id == approval_id
self.approval.incident_id = incident_id
@pytest.mark.asyncio
async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads_back(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
approval_service = _FakeApprovalService(calls)
incident_id = "INC-20260711-BACKUP"
identity: dict[str, str] = {}
async def create_incident(**_kwargs) -> str:
calls.append("incident_create")
return incident_id
async def canonical_readback(value: str, *, project_id: str):
calls.append("incident_canonical_readback")
assert value == incident_id
assert project_id == "awoooi"
return SimpleNamespace(status="investigating", persisted_to_pg=True)
async def dispatch(**kwargs):
calls.append("agent99_dispatch")
assert kwargs["incident_id"] == incident_id
assert kwargs["route_id"] == "agent99:backup_health:BackupCheck"
identity.update({
"incident_id": incident_id,
"run_id": "12345678-1234-5678-9234-567812345678",
"trace_id": "00-11111111111111111111111111111111-2222222222222222-01",
"work_item_id": kwargs["work_item_id"],
})
return {
"status": "dispatched",
"dispatchPerformed": True,
"identity": dict(identity),
}
async def readback(**kwargs):
calls.append("dispatch_receipt_readback")
assert kwargs == {"project_id": "awoooi", "incident_id": incident_id}
return {
"status": "dispatch_accepted_verifier_pending",
"identity": dict(identity),
"dispatch_receipt": {
"kind": "backup_health",
"accepted": True,
"inbox_triggered": True,
},
"verifier": {"status": "pending"},
"runtime_execution_authorized": False,
"runtime_closure_verified": False,
}
monkeypatch.setattr(ingress, "get_approval_service", lambda: approval_service)
monkeypatch.setattr(ingress, "create_incident_for_approval", create_incident)
monkeypatch.setattr(
ingress,
"get_incident_service",
lambda: SimpleNamespace(get_for_readback=canonical_readback),
)
monkeypatch.setattr(ingress, "bridge_alertmanager_to_agent99", dispatch)
monkeypatch.setattr(ingress, "read_agent99_dispatch_receipt", readback)
monkeypatch.setattr(ingress, "record_alertmanager_event", AsyncMock())
result = await ingress.process_backup_restore_alertmanager_signal(
**_signal_kwargs()
)
assert calls.index("owner_create") < calls.index("incident_create")
assert calls.index("incident_create") < calls.index("incident_bind")
assert calls.index("incident_bind") < calls.index("incident_canonical_readback")
assert calls.index("incident_canonical_readback") < calls.index(
"agent99_dispatch"
)
assert calls.index("agent99_dispatch") < calls.index(
"dispatch_receipt_readback"
)
assert result["status"] == "backupcheck_dispatched_verifier_pending"
assert result["receipt_persisted"] is True
assert result["accepted"] is True
assert result["verifier"]["status"] == "pending"
assert result["verifier"]["terminal"] is False
assert result["runtime_execution_authorized"] is False
assert result["runtime_closure_verified"] is False
assert result["production_write_performed"] is False
assert approval_service.created_request.risk_level.value == "low"
assert approval_service.created_request.metadata["telegram_role"] == (
"receipt_projection_only"
)
assert "run_restore" in result["prohibited_actions"]
class _FakeDispatchLedger:
def __init__(self) -> None:
self.receipt = None
self.complete_calls = 0
async def reserve(self, *, identity, payload):
if self.receipt is not None:
return {
"status": "idempotent_waiting_tool",
"claimed": False,
"identity": identity.public_dict(),
"receipt": self.receipt,
}
return {
"status": "reserved",
"claimed": True,
"claim_token": "claim-token-1",
"identity": identity.public_dict(),
"receipt": None,
}
async def complete(self, *, identity, dispatch_receipt, **_kwargs):
self.complete_calls += 1
self.receipt = {
**build_agent99_dispatch_receipt_envelope(
identity=identity,
dispatch_receipt=dispatch_receipt,
),
"receipt_persisted": True,
}
return self.receipt
async def mark_delivery_attempt_started(self, *, identity, **_kwargs):
self.receipt = {
"status": "dispatch_delivery_unknown_reconcile_only",
"identity": identity.public_dict(),
"retry_policy": {"safe_to_retry": False, "reconcile_only": True},
"receipt_persisted": True,
"runtime_closure_verified": False,
}
return self.receipt
@pytest.mark.asyncio
async def test_backup_bridge_postgres_identity_deduplicates_transport(
monkeypatch: pytest.MonkeyPatch,
) -> None:
ledger = _FakeDispatchLedger()
transported: list[dict[str, object]] = []
async def lock(*_args, **_kwargs):
return {"acquired": True, "reason": "acquired"}
monkeypatch.setattr(
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
lambda: ledger,
)
monkeypatch.setattr(
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
lock,
)
monkeypatch.setattr(
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
lambda payload: transported.append(payload)
or {
"schema_version": "agent99_sre_dispatch_receipt_v1",
"status": "accepted_inbox_triggered",
"transport": "relay",
"alert_id": str(payload["id"]),
"kind": str(payload["kind"]),
"accepted": True,
"inbox_triggered": True,
"delivery_certainty": "delivered",
},
)
common = {
"alertname": "BackupCredentialEscrowEvidenceMissing",
"severity": "critical",
"namespace": "awoooi-prod",
"target_resource": "backup_restore",
"message": "backup freshness and escrow evidence missing",
"labels": {"event_type": "backup_restore_escrow_signal"},
"fingerprint": "stable-backup-source-fingerprint",
"project_id": "awoooi",
"incident_id": "INC-20260711-BACKUP",
"approval_id": "12345678-1234-5678-9234-567812345678",
"route_id": "agent99:backup_health:BackupCheck",
"work_item_id": "backupcheck:awoooi:stable",
}
first = await bridge_alertmanager_to_agent99(
alert_id="backup-alert-1",
**common,
)
replay = await bridge_alertmanager_to_agent99(
alert_id="backup-alert-2",
**common,
)
assert first["status"] == "dispatched"
assert replay["status"] == "deduplicated"
assert first["identity"] == replay["identity"]
assert first["dispatchPerformed"] is True
assert replay["dispatchPerformed"] is False
assert ledger.complete_calls == 1
assert len(transported) == 1
assert transported[0]["kind"] == "backup_health"
assert transported[0]["suggestedMode"] == "BackupCheck"
assert transported[0]["controlledApply"] is False
assert "do not run backup/restore" in str(transported[0]["instruction"])