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"])