from __future__ import annotations # ruff: noqa: E402 import os from types import SimpleNamespace from uuid import UUID os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from pydantic import ValidationError from src.api.v1 import agents as agents_api from src.core.config import settings from src.models.agent99_completion import Agent99CompletionCallbackRequest from src.services import agent99_completion_callback as callback_service def payload() -> dict: return { "schema_version": "agent99_completion_callback_v1", "callback_id": "agent99-sre-alert-20260711-resolved", "project_id": "awoooi", "run_id": "sre-alert-20260711", "trace_id": "host_recovery:cold-start-gate:no-host", "work_item_id": "agent99-incident:INC-20260711-001", "alert_id": "INC-20260711-001", "correlation_key": "host_recovery:cold-start-gate:no-host", "source": "agent99-sre-alert-inbox", "mode": "Recover", "outcome_state": "resolved", "controlled_apply": True, "transport_ok": True, "verifier_name": "recover_post_condition_v1", "verifier_passed": True, "source_event_resolved": True, "source_event_resolution_policy": "mode_verifier_can_resolve", "duration_seconds": 9.4, "evidence_ref": "agent99-Recover-20260711-202342.json", "alert_kind": "host_recovery", "alert_service": "cold-start-gate", "alert_host": "192.168.0.110", "telegram": { "sent": True, "suppressed": False, "message_format": "incident_card_zh_tw_v4", "visual_sent": True, "visual_file": "agent99-card-recovery.png", }, "problem": {"occurrences": 2, "verified_resolved": 2}, "occurred_at": "2026-07-11T20:23:51+08:00", } def app_client() -> TestClient: app = FastAPI() app.include_router(agents_api.router, prefix="/api/v1") return TestClient(app) def test_completion_callback_fails_closed_without_server_token(monkeypatch) -> None: monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "") response = app_client().post( "/api/v1/agents/agent99/completion-callback", json=payload(), ) assert response.status_code == 503 assert response.json()["detail"] == "agent99_completion_auth_not_configured" def test_completion_payload_rejects_secret_fields_and_full_paths() -> None: with pytest.raises(ValidationError): Agent99CompletionCallbackRequest.model_validate( {**payload(), "bot_token": "must-not-enter-callback"} ) with pytest.raises(ValidationError): Agent99CompletionCallbackRequest.model_validate( {**payload(), "evidence_ref": r"C:\Wooo\Agent99\evidence\raw.json"} ) def test_completion_callback_rejects_wrong_token(monkeypatch) -> None: monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") response = app_client().post( "/api/v1/agents/agent99/completion-callback", headers={"X-Agent99-Completion-Token": "wrong"}, json=payload(), ) assert response.status_code == 401 assert response.json()["detail"] == "agent99_completion_auth_failed" def test_completion_callback_requires_durable_readback(monkeypatch) -> None: monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") async def fake_record(_payload): return {"ok": False, "durable_readback": False} monkeypatch.setattr(agents_api, "record_agent99_completion_callback", fake_record) response = app_client().post( "/api/v1/agents/agent99/completion-callback", headers={"X-Agent99-Completion-Token": "expected"}, json=payload(), ) assert response.status_code == 503 assert response.json()["detail"] == "agent99_completion_durable_readback_failed" def test_completion_callback_returns_same_trace_receipt(monkeypatch) -> None: monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") async def fake_record(request): return { "schema_version": "agent99_completion_callback_receipt_v1", "ok": True, "durable_readback": True, "callback_id": request.callback_id, "run_id": request.run_id, "trace_id": request.trace_id, "awooop_run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f", } monkeypatch.setattr(agents_api, "record_agent99_completion_callback", fake_record) response = app_client().post( "/api/v1/agents/agent99/completion-callback", headers={"X-Agent99-Completion-Token": "expected"}, json=payload(), ) assert response.status_code == 200 assert response.json()["durable_readback"] is True assert response.json()["callback_id"] == payload()["callback_id"] assert response.json()["trace_id"] == payload()["trace_id"] class FakeOperationRepository: def __init__(self) -> None: self.events: list[tuple[str, dict]] = [] async def append(self, event_type: str, **kwargs): self.events.append((event_type, kwargs)) return SimpleNamespace(id=f"operation-{len(self.events)}") @pytest.mark.asyncio async def test_completion_service_writes_event_run_and_operation_receipts(monkeypatch) -> None: request = Agent99CompletionCallbackRequest.model_validate(payload()) event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c") awooop_run_id = "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f" repository = FakeOperationRepository() readbacks = iter( [ None, { "event_id": str(event_id), "run_id": awooop_run_id, "operation_count": 2, "execution_count": 1, "resolved_count": 1, }, ] ) event_call: dict = {} async def fake_readback(**_kwargs): return next(readbacks) async def fake_record_event(**kwargs): event_call.update(kwargs) return event_id monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback) monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event) monkeypatch.setattr( callback_service, "get_alert_operation_log_repository", lambda: repository, ) receipt = await callback_service.record_agent99_completion_callback(request) assert receipt["durable_readback"] is True assert receipt["awooop_run_id"] == awooop_run_id assert event_call["provider"] == "agent99" assert event_call["stage"] == "resolved" assert [item[0] for item in repository.events] == [ "EXECUTION_COMPLETED", "RESOLVED", ] assert repository.events[0][1]["context"]["stores_secret"] is False assert repository.events[0][1]["context"]["raw_log_stored"] is False @pytest.mark.asyncio async def test_completion_service_is_idempotent(monkeypatch) -> None: request = Agent99CompletionCallbackRequest.model_validate(payload()) event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c") existing = { "event_id": str(event_id), "run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f", "operation_count": 2, "execution_count": 1, "resolved_count": 1, } repository = FakeOperationRepository() event_call: dict = {} async def fake_readback(**_kwargs): return existing async def fake_record_event(**kwargs): event_call.update(kwargs) return event_id monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback) monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event) monkeypatch.setattr( callback_service, "get_alert_operation_log_repository", lambda: repository, ) receipt = await callback_service.record_agent99_completion_callback(request) assert receipt["status"] == "duplicate_verified" assert receipt["duplicate"] is True assert event_call["is_duplicate"] is True assert repository.events == [] @pytest.mark.asyncio async def test_completion_service_repairs_partial_terminal_receipt(monkeypatch) -> None: request = Agent99CompletionCallbackRequest.model_validate(payload()) event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c") before = { "event_id": str(event_id), "run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f", "operation_count": 1, "execution_count": 1, "resolved_count": 0, } after = {**before, "operation_count": 2, "resolved_count": 1} readbacks = iter([before, after]) repository = FakeOperationRepository() async def fake_readback(**_kwargs): return next(readbacks) async def fake_record_event(**_kwargs): return event_id monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback) monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event) monkeypatch.setattr( callback_service, "get_alert_operation_log_repository", lambda: repository, ) receipt = await callback_service.record_agent99_completion_callback(request) assert receipt["durable_readback"] is True assert [item[0] for item in repository.events] == ["RESOLVED"]