2377 lines
77 KiB
Python
2377 lines
77 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
from contextlib import asynccontextmanager
|
|
from dataclasses import replace
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.core import redis_client as redis_client_module
|
|
from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job
|
|
from src.services import awooop_ansible_check_mode_service as service
|
|
from src.services import telegram_gateway as telegram_gateway_module
|
|
from src.services.awooop_ansible_audit_service import (
|
|
build_typed_target_route_generation,
|
|
)
|
|
from src.services.telegram_gateway import TelegramGateway
|
|
from src.workers import ansible_executor_broker as broker
|
|
|
|
DESTINATION_BINDING = "a" * 64
|
|
PROVIDER_DESTINATION_BINDING = "b" * 64
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD = (
|
|
"requested_chat_id_matches_provider_chat_id"
|
|
)
|
|
|
|
|
|
class _MappingResult:
|
|
def __init__(self, row: dict | None = None, scalar=None) -> None:
|
|
self._row = row
|
|
self._scalar = scalar
|
|
|
|
def mappings(self) -> _MappingResult:
|
|
return self
|
|
|
|
def one_or_none(self) -> dict | None:
|
|
return self._row
|
|
|
|
def scalar_one_or_none(self):
|
|
return self._scalar
|
|
|
|
def scalar(self):
|
|
return self._scalar
|
|
|
|
|
|
class _SequenceDB:
|
|
def __init__(self, *results: _MappingResult) -> None:
|
|
self._results = list(results)
|
|
self.statements: list[str] = []
|
|
self.parameters: list[dict] = []
|
|
|
|
async def execute(self, statement, parameters=None):
|
|
self.statements.append(str(statement))
|
|
self.parameters.append(dict(parameters or {}))
|
|
return self._results.pop(0) if self._results else _MappingResult()
|
|
|
|
|
|
class _RowsResult:
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
self._rows = rows
|
|
|
|
def mappings(self) -> _RowsResult:
|
|
return self
|
|
|
|
def all(self) -> list[dict]:
|
|
return self._rows
|
|
|
|
|
|
def _claim() -> service.AnsibleCheckModeClaim:
|
|
typed_target_route = {
|
|
"schema_version": "typed_domain_target_route_v2",
|
|
"route_id": "typed:host_systemd:verified-closure",
|
|
"resolution_status": "resolved",
|
|
"executor": "host_ansible_executor",
|
|
"verifier": "host_runtime_independent_verifier",
|
|
"cross_domain_fallback_allowed": False,
|
|
"allowed_catalog_ids": [
|
|
"ansible:awoooi-auto-repair-canary"
|
|
],
|
|
"allowed_inventory_hosts": ["host_121"],
|
|
"canonical_asset_id": "service:awoooi-auto-repair-canary",
|
|
"target_kind": "host_systemd",
|
|
"risk_class": "medium",
|
|
"controlled_apply_allowed": True,
|
|
"critical_break_glass_required": False,
|
|
}
|
|
generation_candidates = [
|
|
{
|
|
"catalog_id": "ansible:awoooi-auto-repair-canary",
|
|
"catalog_revision": "",
|
|
"canonical_asset_id": "service:awoooi-auto-repair-canary",
|
|
"typed_domain": "host_systemd",
|
|
"inventory_hosts": ["host_121"],
|
|
}
|
|
]
|
|
return service.AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000101",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000100",
|
|
incident_id="INC-20260711-D037E5",
|
|
catalog_id="ansible:awoooi-auto-repair-canary",
|
|
playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml",
|
|
apply_playbook_path=(
|
|
"infra/ansible/playbooks/awoooi-auto-repair-canary.yml"
|
|
),
|
|
inventory_hosts=("host_121",),
|
|
risk_level="medium",
|
|
input_payload={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"approval_id": "00000000-0000-0000-0000-000000000103",
|
|
"candidate_catalog_schema": "typed_domain_router_v2",
|
|
"target_route_generation": build_typed_target_route_generation(
|
|
typed_target_route=typed_target_route,
|
|
executor_candidates=generation_candidates,
|
|
),
|
|
"typed_target_route": typed_target_route,
|
|
"typed_generation_candidates": generation_candidates,
|
|
"cross_domain_fallback_allowed": False,
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
"canonical_asset_id": "service:awoooi-auto-repair-canary",
|
|
"typed_domain": "host_systemd",
|
|
"target_scope_validated": True,
|
|
"catalog_revision": "",
|
|
},
|
|
)
|
|
|
|
|
|
def _verified_result() -> service.AnsibleRunResult:
|
|
return service.AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="",
|
|
duration_ms=25,
|
|
post_verifier_passed=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_apply_replay_rejects_legacy_claim_before_ansible(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
legacy_claim = _claim()
|
|
legacy_claim.input_payload.pop("candidate_catalog_schema")
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_open_failed_apply_retry_row",
|
|
AsyncMock(return_value={"op_id": "legacy-apply-op"}),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_claim_from_apply_operation_row",
|
|
lambda _row: (
|
|
legacy_claim,
|
|
service.AnsibleRunResult(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="failed",
|
|
duration_ms=1,
|
|
),
|
|
),
|
|
)
|
|
runner = AsyncMock(
|
|
side_effect=AssertionError("legacy claim must not touch Ansible")
|
|
)
|
|
monkeypatch.setattr(service, "_run_ansible_command", runner)
|
|
|
|
result = await service.run_failed_apply_check_mode_replay_once(
|
|
timeout_seconds=30,
|
|
)
|
|
|
|
assert result["scanned"] == 1
|
|
assert result["error"] == "failed_apply_replay_typed_scope_not_verified"
|
|
assert result["blockers"] == ["typed_candidate_catalog_schema_required"]
|
|
assert result["runtime_apply_executed"] == 0
|
|
runner.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_apply_replay_rejects_foreign_playbook_before_ansible(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
foreign_claim = replace(
|
|
_claim(),
|
|
playbook_path="infra/ansible/playbooks/foreign-readonly.yml",
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_open_failed_apply_retry_row",
|
|
AsyncMock(return_value={"op_id": "foreign-playbook-apply-op"}),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_claim_from_apply_operation_row",
|
|
lambda _row: (
|
|
foreign_claim,
|
|
service.AnsibleRunResult(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="failed",
|
|
duration_ms=1,
|
|
),
|
|
),
|
|
)
|
|
runner = AsyncMock(
|
|
side_effect=AssertionError("foreign playbook must not touch Ansible")
|
|
)
|
|
monkeypatch.setattr(service, "_run_ansible_command", runner)
|
|
|
|
result = await service.run_failed_apply_check_mode_replay_once(
|
|
timeout_seconds=30,
|
|
)
|
|
|
|
assert result["scanned"] == 1
|
|
assert result["error"] == "failed_apply_replay_typed_scope_not_verified"
|
|
assert result["blockers"] == [
|
|
"typed_claim_check_playbook_catalog_mismatch"
|
|
]
|
|
assert result["runtime_apply_executed"] == 0
|
|
runner.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminal_apply_replaces_queued_approval_truth(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
approval = SimpleNamespace(
|
|
status=service.ApprovalStatus.APPROVED,
|
|
resolved_at=None,
|
|
rejection_reason=None,
|
|
extra_metadata={
|
|
"execution_kind": "host_ansible_check_mode_queued",
|
|
"execution_state": "queued",
|
|
"repair_executed": False,
|
|
"repair_verified": False,
|
|
"incident_closure_allowed": False,
|
|
},
|
|
)
|
|
db = _SequenceDB(_MappingResult(scalar=approval))
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
|
|
|
|
written = await service._finalize_controlled_approval_projection(
|
|
_claim(),
|
|
_verified_result(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert written is True
|
|
assert approval.status == service.ApprovalStatus.EXECUTION_SUCCESS
|
|
assert approval.extra_metadata["execution_kind"] == "controlled_apply"
|
|
assert approval.extra_metadata["execution_state"] == "completed"
|
|
assert approval.extra_metadata["repair_executed"] is True
|
|
assert approval.extra_metadata["repair_verified"] is True
|
|
assert approval.extra_metadata["incident_closure_allowed"] is True
|
|
|
|
|
|
def _controlled_result_payload() -> dict:
|
|
return {
|
|
"chat_id": "sre-chat",
|
|
"text": "CONTROLLED APPLY RESULT INC-20260711-D037E5",
|
|
"_awoooi_canonical_telegram_route": {
|
|
"product_id": "awoooi",
|
|
"signal_family": "incident_lifecycle",
|
|
"severity": "P1",
|
|
},
|
|
"_skip_incident_thread_reply": True,
|
|
"_awooop_source_envelope_extra": {
|
|
"callback_reply": {
|
|
"action": "controlled_apply_result",
|
|
"status": "callback_reply_sent",
|
|
"project_id": "awoooi",
|
|
"automation_run_id": (
|
|
"00000000-0000-0000-0000-000000000102"
|
|
),
|
|
"incident_id": "INC-20260711-D037E5",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def test_apply_result_prefers_safe_playbook_marker_over_stderr_warning() -> None:
|
|
result = service.AnsibleRunResult(
|
|
returncode=2,
|
|
stdout=(
|
|
"TASK [Require successful bounded Compose activation]\n"
|
|
"fatal: [host_188]: FAILED! => {\"msg\": "
|
|
"\"openclaw_compose_activation_failed:bind_source_missing\"}"
|
|
),
|
|
stderr=(
|
|
"[WARNING]: Platform linux on host host_188 is using the "
|
|
"discovered Python interpreter."
|
|
),
|
|
duration_ms=123,
|
|
)
|
|
|
|
status, output, dry_run_result, error = (
|
|
service._build_apply_result_payload(result)
|
|
)
|
|
|
|
assert status == "failed"
|
|
assert error == (
|
|
"openclaw_compose_activation_failed:bind_source_missing"
|
|
)
|
|
assert output["stderr_tail"].startswith("[WARNING]")
|
|
assert dry_run_result["returncode"] == 2
|
|
|
|
|
|
def test_check_mode_result_projects_safe_playbook_failure_contract() -> None:
|
|
result = service.AnsibleRunResult(
|
|
returncode=2,
|
|
stdout=(
|
|
"TASK [Fail closed when a payload base is not hash-compatible]\n"
|
|
"fatal: [host_188]: FAILED! => {\"msg\": "
|
|
"\"openclaw_payload_base_hash_mismatch; "
|
|
"drift_work_item=DRIFT-H188-OPENCLAW-SOURCE-001\"}"
|
|
),
|
|
stderr="[WARNING]: interpreter discovery warning",
|
|
duration_ms=123,
|
|
)
|
|
|
|
status, output, dry_run_result, error = service._build_result_payload(result)
|
|
|
|
assert status == "failed"
|
|
assert error == "openclaw_payload_base_hash_mismatch"
|
|
assert output["failure_class"] == "playbook_contract"
|
|
assert output["failure_reason"] == "openclaw_payload_base_hash_mismatch"
|
|
assert dry_run_result["failure_class"] == "playbook_contract"
|
|
assert dry_run_result["failure_reason"] == "openclaw_payload_base_hash_mismatch"
|
|
|
|
|
|
def test_check_mode_result_projects_fixed_transport_reason_without_raw_output() -> None:
|
|
result = service.AnsibleRunResult(
|
|
returncode=4,
|
|
stdout=(
|
|
"host_188 | UNREACHABLE! => {\"msg\": "
|
|
"\"Failed to connect to the host via ssh: private detail\"}"
|
|
),
|
|
stderr="",
|
|
duration_ms=123,
|
|
)
|
|
|
|
_status, output, dry_run_result, error = service._build_result_payload(result)
|
|
|
|
assert error == "ansible_check_mode_failed_rc_4"
|
|
assert output["failure_class"] == "transport"
|
|
assert output["failure_reason"] == "transport_unreachable"
|
|
assert dry_run_result["failure_reason"] == "transport_unreachable"
|
|
assert "private detail" not in output["failure_reason"]
|
|
|
|
|
|
def test_apply_result_does_not_project_untrusted_stdout_as_error() -> None:
|
|
result = service.AnsibleRunResult(
|
|
returncode=2,
|
|
stdout="fatal: [host_188]: FAILED! => secret-shaped-untrusted-output",
|
|
stderr="safe_transport_failure",
|
|
duration_ms=123,
|
|
)
|
|
|
|
_status, _output, _dry_run_result, error = (
|
|
service._build_apply_result_payload(result)
|
|
)
|
|
|
|
assert error == "safe_transport_failure"
|
|
|
|
|
|
def test_runtime_stage_ids_only_accepts_durable_receipts() -> None:
|
|
stage_ids = service._runtime_stage_ids(
|
|
{
|
|
"runtime_stage_receipts": [
|
|
{
|
|
"stage_id": "mcp_context",
|
|
"durable_receipt": True,
|
|
},
|
|
{
|
|
"stage_id": "telegram_receipt",
|
|
"durable_receipt": False,
|
|
},
|
|
]
|
|
}
|
|
)
|
|
|
|
assert stage_ids == {"mcp_context"}
|
|
|
|
|
|
def test_incident_terminal_writer_normalizes_legacy_non_object_outcome() -> None:
|
|
source = inspect.getsource(service._record_incident_terminal_disposition)
|
|
|
|
assert "jsonb_typeof(CAST(outcome AS jsonb))" in source
|
|
assert "jsonb_build_object(" in source
|
|
assert "'legacy_outcome'" in source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("proposal_executed", "expected_success"),
|
|
[(True, False), (False, None)],
|
|
)
|
|
async def test_incident_terminal_round_trips_negative_outcome_truth(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
proposal_executed: bool,
|
|
expected_success: bool | None,
|
|
) -> None:
|
|
terminal = {
|
|
"schema_version": "ansible_incident_terminal_disposition_v1",
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
"terminal_type": "no_write_replay_failed_waiting_repair_candidate",
|
|
}
|
|
db = _SequenceDB(
|
|
_MappingResult({"proposal_executed": proposal_executed}),
|
|
_MappingResult(
|
|
{
|
|
"incident_status": "MITIGATING",
|
|
"updated_at": None,
|
|
"resolved_at": None,
|
|
"outcome": {
|
|
"automation_terminal": terminal,
|
|
"proposal_executed": proposal_executed,
|
|
"execution_success": expected_success,
|
|
},
|
|
}
|
|
),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
yield db
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
|
|
result = await service._record_incident_terminal_disposition(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
terminal_type=terminal["terminal_type"],
|
|
success_terminal=False,
|
|
retry_op_id=None,
|
|
telegram_receipt_acknowledged=False,
|
|
)
|
|
|
|
assert result is not None
|
|
assert result["proposal_executed"] is proposal_executed
|
|
assert result["execution_success"] is expected_success
|
|
assert db.parameters[1]["execution_success"] is expected_success
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
atomic_writer = AsyncMock()
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
AsyncMock(
|
|
return_value={
|
|
"ready": False,
|
|
"missing": ["telegram_receipt"],
|
|
"receipts": {},
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_commit_verified_incident_closure",
|
|
atomic_writer,
|
|
)
|
|
|
|
result = await service._finalize_verified_apply_closure(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert result == {
|
|
"status": "closure_receipts_pending",
|
|
"closed": False,
|
|
"missing": ["telegram_receipt"],
|
|
}
|
|
atomic_writer.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closure_resolves_only_after_durable_readback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
AsyncMock(return_value={"ready": True, "receipts": {"all": True}}),
|
|
)
|
|
atomic_writer = AsyncMock(
|
|
return_value={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
"incident_resolved": True,
|
|
"closure_receipt_written": True,
|
|
"resolved_lifecycle_written": True,
|
|
"working_memory_projection_ready": True,
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_commit_verified_incident_closure",
|
|
atomic_writer,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_incident_closure_readback",
|
|
AsyncMock(
|
|
return_value={
|
|
"closed": True,
|
|
"missing": [],
|
|
"incident_status": "RESOLVED",
|
|
}
|
|
),
|
|
)
|
|
|
|
result = await service._finalize_verified_apply_closure(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert result["status"] == "controlled_apply_closed"
|
|
assert result["closed"] is True
|
|
atomic_writer.assert_awaited_once()
|
|
assert atomic_writer.await_args.kwargs["closure_prerequisite_count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closure_preserves_durable_terminal_when_projection_is_degraded(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
AsyncMock(return_value={"ready": True, "receipts": {"all": True}}),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_commit_verified_incident_closure",
|
|
AsyncMock(return_value={"working_memory_projection_ready": False}),
|
|
)
|
|
readback = AsyncMock(
|
|
return_value={
|
|
"closed": True,
|
|
"missing": [],
|
|
"incident_status": "RESOLVED",
|
|
}
|
|
)
|
|
monkeypatch.setattr(service, "_read_incident_closure_readback", readback)
|
|
|
|
result = await service._finalize_verified_apply_closure(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert result == {
|
|
"status": "controlled_apply_closed_projection_degraded",
|
|
"closed": True,
|
|
"missing": ["working_memory_terminal_projection"],
|
|
"incident_status": "RESOLVED",
|
|
"working_memory_projection_ready": False,
|
|
"read_model_degraded": True,
|
|
}
|
|
readback.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminal_working_memory_reconciler_projects_durable_rows(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
|
|
updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC)
|
|
db = _SequenceDB(
|
|
_RowsResult(
|
|
[
|
|
{
|
|
"incident_id": "INC-20260711-C50677",
|
|
"incident_status": "RESOLVED",
|
|
"updated_at": updated_at,
|
|
"resolved_at": updated_at,
|
|
},
|
|
{
|
|
"incident_id": "INC-20260711-8A28BC",
|
|
"incident_status": "RESOLVED",
|
|
"updated_at": updated_at,
|
|
"resolved_at": updated_at,
|
|
},
|
|
]
|
|
)
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
yield db
|
|
|
|
project = AsyncMock(side_effect=["updated", "current"])
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_project_incident_terminal_to_working_memory",
|
|
project,
|
|
)
|
|
|
|
result = await service.reconcile_verified_incident_working_memory_once(
|
|
project_id="awoooi",
|
|
window_hours=24,
|
|
limit=6,
|
|
)
|
|
|
|
assert result == {
|
|
"scanned": 2,
|
|
"updated": 1,
|
|
"current": 1,
|
|
"missing": 0,
|
|
"failed": 0,
|
|
"error": None,
|
|
}
|
|
assert project.await_count == 2
|
|
assert db.parameters[0] == {
|
|
"project_id": "awoooi",
|
|
"window_hours": 24,
|
|
"limit": 6,
|
|
"cursor_updated_at": None,
|
|
"cursor_incident_id": "",
|
|
}
|
|
query = db.statements[0]
|
|
assert "working_memory_projection,ready" in query
|
|
assert "source_updated_at_epoch_us" in query
|
|
assert "cursor_updated_at" in query
|
|
assert "ORDER BY incident.updated_at ASC" in query
|
|
assert service._WORKING_MEMORY_PROJECTION_WINDOW_HOURS >= 7 * 24
|
|
service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminal_reconciler_advances_cursor_after_one_row_fails(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
|
|
first_at = datetime(2026, 7, 11, 11, 40, tzinfo=UTC)
|
|
second_at = datetime(2026, 7, 11, 11, 41, tzinfo=UTC)
|
|
db = _SequenceDB(
|
|
_RowsResult(
|
|
[
|
|
{
|
|
"incident_id": "INC-20260711-FAIL01",
|
|
"incident_status": "RESOLVED",
|
|
"updated_at": first_at,
|
|
"resolved_at": first_at,
|
|
},
|
|
{
|
|
"incident_id": "INC-20260711-NEXT01",
|
|
"incident_status": "RESOLVED",
|
|
"updated_at": second_at,
|
|
"resolved_at": second_at,
|
|
},
|
|
]
|
|
)
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
yield db
|
|
|
|
project = AsyncMock(
|
|
side_effect=[RuntimeError("redis unavailable"), "updated"]
|
|
)
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_project_incident_terminal_to_working_memory",
|
|
project,
|
|
)
|
|
|
|
result = await service.reconcile_verified_incident_working_memory_once(
|
|
project_id="awoooi",
|
|
limit=2,
|
|
)
|
|
|
|
assert result["scanned"] == 2
|
|
assert result["failed"] == 1
|
|
assert result["updated"] == 1
|
|
assert project.await_count == 2
|
|
assert service._WORKING_MEMORY_PROJECTION_CURSOR["awoooi"] == (
|
|
second_at,
|
|
"INC-20260711-NEXT01",
|
|
)
|
|
service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_projection_exception_does_not_erase_durable_closure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
terminal = {
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
}
|
|
updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC)
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(
|
|
{
|
|
"incident_status": "RESOLVED",
|
|
"updated_at": updated_at,
|
|
"resolved_at": updated_at,
|
|
"outcome": {
|
|
"automation_terminal": terminal,
|
|
"proposal_executed": True,
|
|
"execution_success": True,
|
|
},
|
|
"closure_receipt_written": True,
|
|
"resolved_lifecycle_written": True,
|
|
}
|
|
),
|
|
)
|
|
transaction_completed = False
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
nonlocal transaction_completed
|
|
yield db
|
|
transaction_completed = True
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_project_incident_terminal_to_working_memory",
|
|
AsyncMock(side_effect=RuntimeError("redis unavailable")),
|
|
)
|
|
|
|
result = await service._commit_verified_incident_closure(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
closure_prerequisite_count=12,
|
|
)
|
|
|
|
assert transaction_completed is True
|
|
assert result is not None
|
|
assert result["incident_resolved"] is True
|
|
assert result["closure_receipt_written"] is True
|
|
assert result["working_memory_projection_status"] == "failed"
|
|
assert result["working_memory_projection_ready"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broker_redis_projection_init_is_fail_soft(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
broker,
|
|
"init_redis_pool",
|
|
AsyncMock(side_effect=RuntimeError("redis unavailable")),
|
|
)
|
|
|
|
assert await broker._init_optional_projection_redis() is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_redis_init_does_not_poison_projection_retry(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
failed_pool = type(
|
|
"_FailedPool",
|
|
(),
|
|
{
|
|
"ping": AsyncMock(side_effect=RuntimeError("redis unavailable")),
|
|
"close": AsyncMock(),
|
|
},
|
|
)()
|
|
monkeypatch.setattr(redis_client_module, "_redis_pool", None)
|
|
monkeypatch.setattr(
|
|
redis_client_module.redis,
|
|
"from_url",
|
|
lambda *_args, **_kwargs: failed_pool,
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="redis unavailable"):
|
|
await redis_client_module.init_redis_pool()
|
|
|
|
assert redis_client_module._redis_pool is None
|
|
failed_pool.close.assert_awaited_once_with()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_atomic_closure_rolls_back_when_bundle_readback_is_missing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(_MappingResult(), _MappingResult(None))
|
|
transaction_rolled_back = False
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
nonlocal transaction_rolled_back
|
|
try:
|
|
yield db
|
|
except Exception:
|
|
transaction_rolled_back = True
|
|
raise
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
|
|
result = await service._commit_verified_incident_closure(
|
|
_claim(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
project_id="awoooi",
|
|
closure_prerequisite_count=12,
|
|
)
|
|
|
|
assert result is None
|
|
assert transaction_rolled_back is True
|
|
atomic_statement = db.statements[1]
|
|
assert "INSERT INTO alert_operation_log" in atomic_statement
|
|
assert "UPDATE automation_operation_log apply" in atomic_statement
|
|
assert "UPDATE incidents incident" in atomic_statement
|
|
assert "WITH target_apply AS" in atomic_statement
|
|
assert "WHEN upper(incident.status::text) = 'CLOSED'" in atomic_statement
|
|
assert db.parameters[1]["apply_op_id_uuid"] == db.parameters[1]["apply_op_id"]
|
|
|
|
|
|
def test_uuid_lookup_and_json_text_receipt_binds_are_type_isolated() -> None:
|
|
mixed_apply_id_queries = (
|
|
service._read_verified_apply_closure_prerequisites,
|
|
service._read_incident_closure_readback,
|
|
service._commit_verified_incident_closure,
|
|
service._verify_retry_terminal_projection_readback,
|
|
)
|
|
|
|
for query_owner in mixed_apply_id_queries:
|
|
source = inspect.getsource(query_owner)
|
|
assert "CAST(:apply_op_id_uuid AS uuid)" in source
|
|
assert "CAST(:apply_op_id AS uuid)" not in source
|
|
assert '"apply_op_id_uuid": apply_op_id' in source
|
|
|
|
retry_source = inspect.getsource(
|
|
service._verify_retry_terminal_projection_readback
|
|
)
|
|
assert "CAST(:replay_op_id_uuid AS uuid)" in retry_source
|
|
assert "CAST(:replay_op_id AS uuid)" not in retry_source
|
|
assert '"replay_op_id_uuid": replay_op_id' in retry_source
|
|
|
|
|
|
def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None:
|
|
predicate = candidate_job.verified_ansible_candidate_terminal_sql(
|
|
"candidate"
|
|
)
|
|
recorder_source = inspect.getsource(
|
|
candidate_job.record_ansible_decision_audit
|
|
)
|
|
|
|
assert "direct_terminal.status = 'dry_run'" in predicate
|
|
assert "check_mode.parent_op_id = candidate.op_id" in predicate
|
|
assert "apply.parent_op_id = check_mode.op_id" in predicate
|
|
assert "replay.parent_op_id = apply.op_id" in predicate
|
|
assert "replay.status IN ('success', 'failed')" in predicate
|
|
assert (
|
|
"replay_check.parent_op_id = failed_check.parent_op_id"
|
|
in predicate
|
|
)
|
|
assert "replay_of_check_mode_op_id" in predicate
|
|
assert "failed_check.parent_op_id = candidate.op_id" in predicate
|
|
assert "failed_check.op_id = CAST(" in predicate
|
|
assert predicate.count("failed_check.op_id::text") == 1
|
|
assert "verified_no_write_terminal" in predicate
|
|
assert "repository_readback_verified" in predicate
|
|
assert "{detail,execution_success}" in predicate
|
|
assert "{detail,telegram_receipt_acknowledged}" in predicate
|
|
assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in (
|
|
recorder_source
|
|
)
|
|
assert "target_route_generation" in recorder_source
|
|
assert "candidate_lane_lock_key" in recorder_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_projection_replay_sends_missing_receipt_without_reapplying(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
closure_coverage = {
|
|
"telegram_receipt": False,
|
|
"candidate": True,
|
|
"check_mode": True,
|
|
"controlled_apply": True,
|
|
"post_apply_verifier": True,
|
|
"auto_repair_execution_receipt": True,
|
|
"km_playbook_writeback": True,
|
|
"rag_writeback": True,
|
|
"mcp_context": True,
|
|
"playbook_trust": True,
|
|
"timeline_projection": True,
|
|
"approval_projection": True,
|
|
"execution_lifecycle": True,
|
|
"required_runtime_stage_receipts": True,
|
|
}
|
|
readback = AsyncMock(
|
|
side_effect=[
|
|
{"receipts": closure_coverage},
|
|
{"receipts": {"telegram_receipt": True}},
|
|
]
|
|
)
|
|
telegram_sender = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_finalize_controlled_approval_projection",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_append_alert_lifecycle_receipt",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
readback,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_send_controlled_apply_telegram_receipt",
|
|
telegram_sender,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_finalize_verified_apply_closure",
|
|
AsyncMock(return_value={"status": "controlled_apply_closed", "closed": True}),
|
|
)
|
|
|
|
result = await service._reconcile_verified_apply_closure_projections(
|
|
_claim(),
|
|
_verified_result(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
writeback={
|
|
"verification_passed": True,
|
|
"verification_result": "success",
|
|
"verification": True,
|
|
"learning": True,
|
|
},
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert result["closed"] is True
|
|
assert result["runtime_apply_executed"] is False
|
|
telegram_sender.assert_awaited_once()
|
|
assert telegram_sender.await_args.kwargs["closure_receipts"] == (
|
|
closure_coverage
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_chain_final_receipt_projects_all_learning_evidence(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = AsyncMock()
|
|
gateway.send_controlled_apply_result_receipt.return_value = {
|
|
"ok": True,
|
|
"_awooop_outbound_mirror_acknowledged": True,
|
|
}
|
|
monkeypatch.setattr(
|
|
"src.services.telegram_gateway.get_telegram_gateway",
|
|
lambda: gateway,
|
|
)
|
|
base = _claim()
|
|
claim = replace(
|
|
base,
|
|
catalog_id="ansible:110-alertmanager-delivery-recovery",
|
|
playbook_path=(
|
|
"infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml"
|
|
),
|
|
apply_playbook_path=(
|
|
"infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml"
|
|
),
|
|
input_payload={
|
|
**base.input_payload,
|
|
"alert_category": "alert_chain_health",
|
|
},
|
|
)
|
|
|
|
acknowledged = await service._send_controlled_apply_telegram_receipt(
|
|
claim,
|
|
_verified_result(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
writeback={
|
|
"verification_result": "success",
|
|
"verification": True,
|
|
"learning": True,
|
|
"rag_writeback": True,
|
|
"trust_learning": True,
|
|
},
|
|
closure_receipts={
|
|
"km_playbook_writeback": True,
|
|
"rag_writeback": True,
|
|
"mcp_context": True,
|
|
"playbook_trust": True,
|
|
},
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert acknowledged is True
|
|
call = gateway.send_controlled_apply_result_receipt.await_args.kwargs
|
|
assert call["alert_category"] == "alert_chain_health"
|
|
assert call["verifier_written"] is True
|
|
assert call["learning_written"] is True
|
|
assert call["rag_written"] is True
|
|
assert call["mcp_context_ready"] is True
|
|
assert call["playbook_trust_written"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_projection_stays_open_until_durable_telegram_readback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
lifecycle = AsyncMock(return_value=True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_finalize_controlled_approval_projection",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
monkeypatch.setattr(service, "_append_alert_lifecycle_receipt", lifecycle)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
AsyncMock(
|
|
side_effect=[
|
|
{"receipts": {"telegram_receipt": False}},
|
|
{"receipts": {"telegram_receipt": False}},
|
|
]
|
|
),
|
|
)
|
|
telegram_sender = AsyncMock(return_value=False)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_send_controlled_apply_telegram_receipt",
|
|
telegram_sender,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_finalize_verified_apply_closure",
|
|
AsyncMock(
|
|
return_value={
|
|
"status": "closure_receipts_pending",
|
|
"closed": False,
|
|
"missing": ["telegram_receipt"],
|
|
}
|
|
),
|
|
)
|
|
|
|
result = await service._reconcile_verified_apply_closure_projections(
|
|
_claim(),
|
|
_verified_result(),
|
|
apply_op_id="00000000-0000-0000-0000-000000000104",
|
|
writeback={
|
|
"verification_passed": True,
|
|
"verification_result": "success",
|
|
"verification": True,
|
|
"learning": True,
|
|
},
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert result["closed"] is False
|
|
assert result["telegram_receipt_acknowledged"] is False
|
|
assert result["telegram_lifecycle_written"] is False
|
|
assert lifecycle.await_count == 1
|
|
telegram_sender.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_controlled_result_pending_reservation_blocks_duplicate_provider_send(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
class _Response:
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict:
|
|
return {
|
|
"ok": True,
|
|
"result": {
|
|
"message_id": 456,
|
|
"chat": {"id": "sre-chat"},
|
|
},
|
|
}
|
|
|
|
class _Client:
|
|
def __init__(self) -> None:
|
|
self.post_count = 0
|
|
|
|
async def post(self, _url, json):
|
|
self.post_count += 1
|
|
return _Response()
|
|
|
|
gateway = TelegramGateway()
|
|
monkeypatch.setattr(
|
|
telegram_gateway_module.settings,
|
|
"SRE_GROUP_CHAT_ID",
|
|
"sre-chat",
|
|
)
|
|
gateway._initialized = True
|
|
client = _Client()
|
|
gateway._http_client = client
|
|
reserve = AsyncMock(
|
|
side_effect=[
|
|
{
|
|
"status": "reserved",
|
|
"run_id": "00000000-0000-0000-0000-000000000201",
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
},
|
|
{
|
|
"status": "pending_unknown",
|
|
"run_id": "00000000-0000-0000-0000-000000000201",
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
"durable_reservation_committed": True,
|
|
"provider_retry_blocked": True,
|
|
},
|
|
]
|
|
)
|
|
finalize = AsyncMock(return_value=False)
|
|
monkeypatch.setattr(
|
|
gateway,
|
|
"_reserve_controlled_apply_result_outbound",
|
|
reserve,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway,
|
|
"_finalize_controlled_apply_result_outbound",
|
|
finalize,
|
|
)
|
|
|
|
first = await gateway._send_request(
|
|
"sendMessage",
|
|
_controlled_result_payload(),
|
|
)
|
|
second = await gateway._send_request(
|
|
"sendMessage",
|
|
_controlled_result_payload(),
|
|
)
|
|
|
|
assert client.post_count == 1
|
|
assert first["_awooop_provider_send_performed"] is True
|
|
assert first["_awooop_outbound_mirror_acknowledged"] is False
|
|
assert second["ok"] is False
|
|
assert second["_awooop_provider_send_performed"] is False
|
|
assert second["_awooop_delivery_status"] == "pending_unknown"
|
|
assert finalize.await_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_durable_reconcile_finalizes_exact_pending_row_without_provider_send(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
pending_row = {
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
"send_status": "pending",
|
|
"provider_message_id": None,
|
|
"destination_binding": DESTINATION_BINDING,
|
|
"visual_requested": True,
|
|
}
|
|
sent_row = {
|
|
**pending_row,
|
|
"send_status": "sent",
|
|
"provider_message_id": "456",
|
|
"destination_binding": DESTINATION_BINDING,
|
|
"provider_destination_binding": PROVIDER_DESTINATION_BINDING,
|
|
"provider_destination_verification_method": (
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
"provider_destination_identity_verified": True,
|
|
}
|
|
db = _SequenceDB(
|
|
_MappingResult(row=pending_row),
|
|
_MappingResult(),
|
|
_MappingResult(),
|
|
_MappingResult(
|
|
row={"send_status": "sent", "provider_message_id": "456"}
|
|
),
|
|
_MappingResult(row=sent_row),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
gateway = TelegramGateway()
|
|
gateway._http_client = AsyncMock()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "agent99-lifecycle-exact-readback",
|
|
"incident_id": "INC-20260716-AG99ACK",
|
|
"apply_op_id": "recovered|info",
|
|
"delivery_kind": "agent99_lifecycle",
|
|
}
|
|
|
|
result = await gateway._reconcile_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
expected_provider_message_id="456",
|
|
expected_destination_binding=DESTINATION_BINDING,
|
|
expected_provider_destination_binding=PROVIDER_DESTINATION_BINDING,
|
|
expected_provider_destination_verification_method=(
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
)
|
|
|
|
assert result == {
|
|
"status": "sent",
|
|
"run_id": str(
|
|
telegram_gateway_module._controlled_apply_result_delivery_run_id(
|
|
identity
|
|
)
|
|
),
|
|
"message_id": pending_row["message_id"],
|
|
"provider_message_id": "456",
|
|
"destination_binding": DESTINATION_BINDING,
|
|
"provider_destination_binding": PROVIDER_DESTINATION_BINDING,
|
|
"provider_destination_verification_method": (
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
"provider_destination_identity_verified": True,
|
|
"visual_requested": True,
|
|
}
|
|
gateway._http_client.post.assert_not_awaited()
|
|
assert len(db.statements) == 5
|
|
assert "{callback_reply,automation_run_id}" in db.statements[0]
|
|
assert "provider_destination_binding" in db.statements[0]
|
|
assert "provider_destination_verification_method" in db.statements[0]
|
|
assert "provider_destination_identity_verified" in db.statements[0]
|
|
assert db.parameters[0]["automation_run_id"] == identity["automation_run_id"]
|
|
assert db.parameters[0]["incident_id"] == identity["incident_id"]
|
|
assert db.parameters[0]["apply_op_id"] == identity["apply_op_id"]
|
|
assert "telegram-destination-pin:" in db.parameters[2][
|
|
"destination_lock_key"
|
|
]
|
|
assert "UPDATE awooop_outbound_message" in db.statements[3]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_controlled_result_finalize_backfills_missing_destination_binding(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(),
|
|
_MappingResult(row={"send_status": "sent", "provider_message_id": "456"}),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
gateway = TelegramGateway()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "agent99-lifecycle-missing-binding",
|
|
"incident_id": "INC-20260717-WAZUHACK",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
"delivery_kind": "controlled_apply_result",
|
|
}
|
|
|
|
finalized = await gateway._finalize_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
reservation={
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
"run_id": str(
|
|
telegram_gateway_module._controlled_apply_result_delivery_run_id(
|
|
identity
|
|
)
|
|
),
|
|
},
|
|
provider_message_id="456",
|
|
expected_destination_binding=DESTINATION_BINDING,
|
|
provider_destination_binding=PROVIDER_DESTINATION_BINDING,
|
|
provider_destination_verification_method=(
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
)
|
|
|
|
assert finalized is True
|
|
update_sql = db.statements[2]
|
|
assert "{canonical_route_receipt,destination_binding}" in update_sql
|
|
assert "COALESCE(" in update_sql
|
|
assert "IN ('', :expected_destination_binding)" in update_sql
|
|
assert "CAST(" in update_sql
|
|
assert ":expected_destination_binding\n" in update_sql
|
|
assert "AS text" in update_sql
|
|
assert ":expected_destination_binding::text" not in update_sql
|
|
assert "NOT EXISTS" in update_sql
|
|
assert "provider_destination_binding" in update_sql
|
|
assert db.parameters[2]["expected_destination_binding"] == (
|
|
DESTINATION_BINDING
|
|
)
|
|
assert db.parameters[2]["provider_destination_binding"] == (
|
|
PROVIDER_DESTINATION_BINDING
|
|
)
|
|
assert db.parameters[2][
|
|
"provider_destination_verification_method"
|
|
] == PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_durable_reconcile_rejects_provider_message_identity_mismatch(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(
|
|
row={
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
"send_status": "sent",
|
|
"provider_message_id": "456",
|
|
"visual_requested": False,
|
|
}
|
|
)
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
gateway = TelegramGateway()
|
|
gateway._http_client = AsyncMock()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "agent99-lifecycle-provider-mismatch",
|
|
"incident_id": "INC-20260716-AG99ACK",
|
|
"apply_op_id": "recovered|info",
|
|
"delivery_kind": "agent99_lifecycle",
|
|
}
|
|
|
|
result = await gateway._reconcile_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
expected_provider_message_id="999",
|
|
)
|
|
|
|
assert result["status"] == "provider_message_id_mismatch"
|
|
assert result["provider_message_id"] == "456"
|
|
assert len(db.statements) == 1
|
|
gateway._http_client.post.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_controlled_result_reuses_durable_sent_reservation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
monkeypatch.setattr(
|
|
telegram_gateway_module.settings,
|
|
"SRE_GROUP_CHAT_ID",
|
|
"sre-chat",
|
|
)
|
|
gateway._initialized = True
|
|
client = AsyncMock()
|
|
gateway._http_client = client
|
|
destination_binding = (
|
|
telegram_gateway_module._telegram_destination_binding("sre-chat")
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway,
|
|
"_reserve_controlled_apply_result_outbound",
|
|
AsyncMock(
|
|
return_value={
|
|
"status": "sent",
|
|
"run_id": "00000000-0000-0000-0000-000000000201",
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
"provider_message_id": "456",
|
|
"destination_binding": destination_binding,
|
|
"provider_destination_binding": destination_binding,
|
|
"provider_destination_verification_method": (
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
"provider_destination_identity_verified": True,
|
|
}
|
|
),
|
|
)
|
|
|
|
result = await gateway._send_request(
|
|
"sendMessage",
|
|
_controlled_result_payload(),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["_awooop_outbound_mirror_acknowledged"] is True
|
|
assert result["_awooop_provider_send_performed"] is False
|
|
assert result["_awooop_delivery_context"] == {
|
|
"destination_binding": destination_binding,
|
|
"provider_destination_binding": destination_binding,
|
|
"provider_destination_verification_method": (
|
|
PROVIDER_DESTINATION_VERIFICATION_METHOD
|
|
),
|
|
"provider_destination_identity_verified": True,
|
|
"destination_binding_verified": True,
|
|
"durable_exact_row_readback": True,
|
|
}
|
|
client.post.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_write_result_reuses_durable_suppression_without_provider_send(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
monkeypatch.setattr(
|
|
telegram_gateway_module.settings,
|
|
"SRE_GROUP_CHAT_ID",
|
|
"sre-chat",
|
|
)
|
|
gateway._initialized = True
|
|
client = AsyncMock()
|
|
gateway._http_client = client
|
|
monkeypatch.setattr(
|
|
gateway,
|
|
"_reserve_controlled_apply_result_outbound",
|
|
AsyncMock(
|
|
return_value={
|
|
"status": "suppressed",
|
|
"run_id": "00000000-0000-0000-0000-000000000201",
|
|
"message_id": "00000000-0000-0000-0000-000000000202",
|
|
}
|
|
),
|
|
)
|
|
|
|
result = await gateway._send_request(
|
|
"sendMessage",
|
|
_controlled_result_payload(),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["_awooop_outbound_mirror_acknowledged"] is True
|
|
assert result["_awooop_provider_send_performed"] is False
|
|
assert result["_awooop_notification_suppressed"] is True
|
|
assert result["_awooop_delivery_status"] == "suppressed"
|
|
client.post.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_write_group_reservation_persists_shadow_suppression(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(row=None),
|
|
_MappingResult(scalar="00000000-0000-0000-0000-000000000299"),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
record_outbound = AsyncMock(
|
|
return_value="00000000-0000-0000-0000-000000000298"
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.db.base.get_db_context",
|
|
fake_get_db_context,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.channel_hub.record_outbound_message",
|
|
record_outbound,
|
|
)
|
|
gateway = TelegramGateway()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"incident_id": "INC-20260711-D037E5",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
}
|
|
source_extra = {
|
|
"callback_reply": {
|
|
"action": "controlled_apply_result",
|
|
"execution_kind": "no_write_replay",
|
|
**identity,
|
|
},
|
|
"notification_policy": {
|
|
"failure_fingerprint": "deadbeefdeadbeefdeadbeef",
|
|
"digest_window_minutes": 30,
|
|
"disposition": "send_or_suppress",
|
|
},
|
|
}
|
|
|
|
reservation = await gateway._reserve_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
payload={"chat_id": "sre-chat", "text": "AI repair digest"},
|
|
source_envelope_extra=source_extra,
|
|
)
|
|
|
|
assert reservation["status"] == "suppressed"
|
|
assert reservation["message_id"] == (
|
|
"00000000-0000-0000-0000-000000000298"
|
|
)
|
|
assert "send_status IN ('pending', 'sent')" in db.statements[2]
|
|
assert db.parameters[0]["lock_key"].endswith(
|
|
":deadbeefdeadbeefdeadbeef"
|
|
)
|
|
record_kwargs = record_outbound.await_args.kwargs
|
|
assert record_kwargs["send_status"] == "shadow"
|
|
assert record_kwargs["provider_message_id"] is None
|
|
assert record_kwargs["is_shadow"] is True
|
|
assert record_kwargs["source_envelope"]["notification_policy"][
|
|
"disposition"
|
|
] == "suppressed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_historical_no_write_backfill_is_provider_silent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(row=None),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
record_outbound = AsyncMock(
|
|
return_value="00000000-0000-0000-0000-000000000398"
|
|
)
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
monkeypatch.setattr(
|
|
"src.services.channel_hub.record_outbound_message",
|
|
record_outbound,
|
|
)
|
|
gateway = TelegramGateway()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000302",
|
|
"incident_id": "INC-20260711-HISTORICAL",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000304",
|
|
}
|
|
source_extra = {
|
|
"callback_reply": {
|
|
"action": "controlled_apply_result",
|
|
"execution_kind": "no_write_replay",
|
|
**identity,
|
|
},
|
|
"notification_policy": {
|
|
"failure_fingerprint": "feedfacefeedfacefeedface",
|
|
"digest_window_minutes": 30,
|
|
"disposition": "send_or_suppress",
|
|
"provider_delivery": "shadow_only",
|
|
},
|
|
}
|
|
|
|
reservation = await gateway._reserve_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
payload={"chat_id": "sre-chat", "text": "historical repair digest"},
|
|
source_envelope_extra=source_extra,
|
|
)
|
|
|
|
assert reservation["status"] == "suppressed"
|
|
assert len(db.statements) == 2
|
|
assert all(
|
|
"send_status IN ('pending', 'sent')" not in sql
|
|
for sql in db.statements
|
|
)
|
|
record_kwargs = record_outbound.await_args.kwargs
|
|
assert record_kwargs["send_status"] == "shadow"
|
|
assert record_kwargs["provider_message_id"] is None
|
|
policy = record_kwargs["source_envelope"]["notification_policy"]
|
|
assert policy["disposition"] == "suppressed"
|
|
assert policy["provider_send_performed"] is False
|
|
assert policy["suppression_reason"] == "historical_projection_backfill"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_historical_verified_repair_is_provider_silent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(row=None),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
record_outbound = AsyncMock(
|
|
return_value="00000000-0000-0000-0000-000000000498"
|
|
)
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
monkeypatch.setattr(
|
|
"src.services.channel_hub.record_outbound_message",
|
|
record_outbound,
|
|
)
|
|
gateway = TelegramGateway()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000402",
|
|
"incident_id": "INC-20260711-HISTORICAL-SUCCESS",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000404",
|
|
}
|
|
source_extra = {
|
|
"callback_reply": {
|
|
"action": "controlled_apply_result",
|
|
"execution_kind": "controlled_apply",
|
|
**identity,
|
|
},
|
|
"notification_policy": {
|
|
"failure_fingerprint": "",
|
|
"disposition": "suppressed",
|
|
"provider_delivery": "shadow_only",
|
|
},
|
|
}
|
|
|
|
reservation = await gateway._reserve_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
payload={"chat_id": "sre-chat", "text": "historical repair closed"},
|
|
source_envelope_extra=source_extra,
|
|
)
|
|
|
|
assert reservation["status"] == "suppressed"
|
|
assert len(db.statements) == 2
|
|
record_kwargs = record_outbound.await_args.kwargs
|
|
assert record_kwargs["send_status"] == "shadow"
|
|
assert record_kwargs["provider_message_id"] is None
|
|
policy = record_kwargs["source_envelope"]["notification_policy"]
|
|
assert policy["suppression_reason"] == "historical_projection_backfill"
|
|
assert policy["provider_send_performed"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_reservation_uses_committed_context_as_durable_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(row=None),
|
|
)
|
|
context_exited = False
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
nonlocal context_exited
|
|
yield db
|
|
context_exited = True
|
|
|
|
record_outbound = AsyncMock(
|
|
return_value="00000000-0000-0000-0000-000000000598"
|
|
)
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context)
|
|
monkeypatch.setattr(
|
|
"src.services.channel_hub.record_outbound_message",
|
|
record_outbound,
|
|
)
|
|
gateway = TelegramGateway()
|
|
identity = {
|
|
"project_id": "awoooi",
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000502",
|
|
"incident_id": "INC-20260715-RESERVATION",
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000504",
|
|
}
|
|
|
|
reservation = await gateway._reserve_controlled_apply_result_outbound(
|
|
identity=identity,
|
|
payload={"chat_id": "sre-chat", "text": "committed reservation"},
|
|
source_envelope_extra={
|
|
"callback_reply": {
|
|
"action": "controlled_apply_result",
|
|
**identity,
|
|
}
|
|
},
|
|
)
|
|
|
|
assert context_exited is True
|
|
assert reservation["status"] == "reserved"
|
|
assert reservation["durable_reservation_committed"] is True
|
|
assert len(db.statements) == 2
|
|
record_outbound.assert_awaited_once()
|
|
|
|
|
|
def test_controlled_result_uses_durable_reservation_and_single_reconcile_entry() -> None:
|
|
reserve_source = inspect.getsource(
|
|
TelegramGateway._reserve_controlled_apply_result_outbound
|
|
)
|
|
finalize_source = inspect.getsource(
|
|
TelegramGateway._finalize_controlled_apply_result_outbound
|
|
)
|
|
gateway_source = inspect.getsource(TelegramGateway._send_request)
|
|
live_apply_source = inspect.getsource(service.run_controlled_apply_for_claim)
|
|
|
|
assert "pg_advisory_xact_lock" in reserve_source
|
|
assert "record_outbound_message" in reserve_source
|
|
assert 'send_status="pending"' in reserve_source
|
|
assert 'send_status="shadow"' in reserve_source
|
|
assert "failure_fingerprint" in reserve_source
|
|
assert "digest_window_minutes" in reserve_source
|
|
assert "notification_suppressed_deduplicated" in reserve_source
|
|
assert "durable_reservation_committed" in reserve_source
|
|
assert "provider_retry_blocked" in reserve_source
|
|
assert "reused_pending_reservation" not in reserve_source
|
|
assert "reservation_readback_failed" not in reserve_source
|
|
assert "UPDATE awooop_outbound_message" in finalize_source
|
|
assert "_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS" in finalize_source
|
|
assert gateway_source.index(
|
|
"_reserve_controlled_apply_result_outbound"
|
|
) < gateway_source.index("_http_client.post")
|
|
assert (
|
|
"telegram_receipt_sent = await _send_controlled_apply_telegram_receipt"
|
|
not in live_apply_source
|
|
)
|
|
assert live_apply_source.count(
|
|
"_reconcile_verified_apply_closure_projections"
|
|
) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_telegram_readback_binds_exact_apply_operation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = _SequenceDB(_MappingResult(scalar=True))
|
|
|
|
@asynccontextmanager
|
|
async def fake_get_db_context(_project_id):
|
|
yield db
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
|
|
apply_op_id = "00000000-0000-0000-0000-000000000104"
|
|
|
|
acknowledged = await service._retry_telegram_receipt_acknowledged(
|
|
_claim(),
|
|
apply_op_id=apply_op_id,
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert acknowledged is True
|
|
assert "{callback_reply,apply_op_id}" in db.statements[0]
|
|
assert db.parameters[0]["apply_op_id"] == apply_op_id
|
|
|
|
|
|
def test_all_terminal_telegram_readbacks_require_exact_apply_operation() -> None:
|
|
closure_source = inspect.getsource(
|
|
service._read_verified_apply_closure_prerequisites
|
|
)
|
|
retry_source = inspect.getsource(
|
|
service._retry_telegram_receipt_acknowledged
|
|
)
|
|
retry_receipt_source = inspect.getsource(
|
|
service._record_retry_runtime_stage_receipt
|
|
)
|
|
projection_source = inspect.getsource(
|
|
service._load_missing_retry_terminal_projection_rows
|
|
)
|
|
|
|
assert closure_source.count("{callback_reply,apply_op_id}") == 1
|
|
assert retry_source.count("{callback_reply,apply_op_id}") == 1
|
|
assert retry_receipt_source.count("{callback_reply,apply_op_id}") == 1
|
|
assert projection_source.count("{callback_reply,apply_op_id}") == 2
|
|
assert projection_source.count("= apply.op_id::text") >= 2
|
|
assert "coalesce(\n outbound.source_envelope #>>\n '{callback_reply,apply_op_id}'" not in projection_source
|
|
|
|
|
|
def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -> None:
|
|
proposal = candidate_job._build_backfill_proposal(
|
|
{
|
|
"alertname": "AwoooPAutoRepairCanaryT16",
|
|
"severity": "medium",
|
|
"approval_id": "00000000-0000-0000-0000-000000000103",
|
|
}
|
|
)
|
|
query_source = inspect.getsource(
|
|
candidate_job._fetch_missing_candidate_incidents
|
|
)
|
|
|
|
assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103"
|
|
assert "approval_records" in query_source
|
|
assert "verified_ansible_candidate_terminal_sql" in query_source
|
|
assert "current_recurrence" in query_source
|
|
assert "target_route_generation" in query_source
|
|
assert "latest_candidate_is_terminal" in query_source
|
|
assert "preflight_ansible_candidate_generation" in inspect.getsource(
|
|
candidate_job.enqueue_missing_ansible_candidates_once
|
|
)
|
|
assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in query_source
|
|
|
|
|
|
def test_retry_projection_only_repairs_latest_candidate_run() -> None:
|
|
source = inspect.getsource(
|
|
service._load_missing_retry_terminal_projection_rows
|
|
)
|
|
|
|
assert "source_candidate.op_id = source_check.parent_op_id" in source
|
|
assert "newer_candidate.created_at," in source
|
|
assert "source_candidate.created_at," in source
|
|
assert "repair_candidate_controlled_queue" in source
|
|
assert "retry_receipt.receipt IS NULL" in source
|
|
assert "controlled_apply_result" in source
|
|
assert "LEFT JOIN LATERAL" in source
|
|
assert "receipt.value ->> 'automation_run_id'" in source
|
|
assert "upper(incident.status::text)" in source
|
|
assert "ansible_incident_terminal_disposition_v1" in source
|
|
assert "'proposal_executed'" in source
|
|
assert "'execution_success'" in source
|
|
assert "verification_result IN ('failed', 'timeout')" in source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broker_reconciles_receipts_then_continues_claiming_fresh_work(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
retry_replayer = AsyncMock(
|
|
return_value={
|
|
"scanned": 0,
|
|
"replayed": 0,
|
|
"check_mode_passed": 0,
|
|
"check_mode_failed": 0,
|
|
"runtime_stage_receipt_written": 0,
|
|
"blockers": [],
|
|
"error": None,
|
|
}
|
|
)
|
|
fresh_claimer = AsyncMock(return_value=[])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_expire_stale_ansible_execution_capabilities",
|
|
AsyncMock(return_value=0),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_retry_terminal_projections_once",
|
|
AsyncMock(
|
|
return_value={
|
|
"scanned": 0,
|
|
"written": 0,
|
|
"incident_receipt_written": 0,
|
|
"lifecycle_written": 0,
|
|
"verified": 0,
|
|
"runtime_apply_executed": False,
|
|
"error": None,
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_auto_repair_execution_receipts_once",
|
|
AsyncMock(
|
|
return_value={
|
|
"scanned": 1,
|
|
"written": 1,
|
|
"incident_closure_written": 1,
|
|
"telegram_receipt_acknowledged": 1,
|
|
"error": None,
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"run_failed_apply_check_mode_replay_once",
|
|
retry_replayer,
|
|
)
|
|
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_stale_pending_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_catalog_drift_failed_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer)
|
|
|
|
result = await service.run_pending_check_modes_once(limit=1)
|
|
|
|
assert result["claimed"] == 0
|
|
assert result["repair_receipt_backfill_priority_tick"] is True
|
|
assert result["repair_receipt_closure_written"] == 1
|
|
retry_replayer.assert_awaited_once()
|
|
fresh_claimer.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broker_errors_do_not_starve_unrelated_fresh_claims(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_expire_stale_ansible_execution_capabilities",
|
|
AsyncMock(return_value=0),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_auto_repair_execution_receipts_once",
|
|
AsyncMock(
|
|
return_value={
|
|
"scanned": 3,
|
|
"written": 0,
|
|
"incident_closure_written": 0,
|
|
"error": "legacy_projection_write_failed",
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"run_failed_apply_check_mode_replay_once",
|
|
AsyncMock(
|
|
return_value={
|
|
"scanned": 1,
|
|
"replayed": 0,
|
|
"blockers": ["foreign_retry_already_claimed"],
|
|
"error": "foreign_retry_reconstruction_failed",
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_stale_pending_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_catalog_drift_failed_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
fresh_claimer = AsyncMock(return_value=[])
|
|
monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer)
|
|
|
|
result = await service.run_pending_check_modes_once(limit=1)
|
|
|
|
assert result["repair_receipt_backfill_error"] == (
|
|
"legacy_projection_write_failed"
|
|
)
|
|
assert result["failed_apply_retry_error"] == (
|
|
"foreign_retry_reconstruction_failed"
|
|
)
|
|
assert result["failed_apply_retry_blockers"] == [
|
|
"foreign_retry_already_claimed"
|
|
]
|
|
fresh_claimer.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
row = {
|
|
"op_id": "00000000-0000-0000-0000-000000000104",
|
|
"replay_op_id": "00000000-0000-0000-0000-000000000105",
|
|
"replay_status": "success",
|
|
"replay_output": {"returncode": 0},
|
|
"replay_dry_run_result": {"timed_out": False},
|
|
"replay_duration_ms": 18,
|
|
"retry_receipt_present": False,
|
|
"terminal_type": "no_write_replay_passed_waiting_repair_candidate",
|
|
"telegram_receipt_acknowledged": False,
|
|
}
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_missing_retry_terminal_projection_rows",
|
|
AsyncMock(return_value=[row]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_claim_from_apply_operation_row",
|
|
lambda _row: (
|
|
_claim(),
|
|
service.AnsibleRunResult(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="",
|
|
duration_ms=0,
|
|
),
|
|
),
|
|
)
|
|
terminal_writer = AsyncMock(
|
|
return_value={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000102",
|
|
"apply_op_id": row["op_id"],
|
|
"retry_op_id": row["replay_op_id"],
|
|
"terminal_type": row["terminal_type"],
|
|
"repository_readback_verified": True,
|
|
}
|
|
)
|
|
receipt_writer = AsyncMock(return_value=True)
|
|
lifecycle_writer = AsyncMock(return_value=True)
|
|
verifier = AsyncMock(return_value=True)
|
|
telegram_sender = AsyncMock(return_value=True)
|
|
telegram_readback = AsyncMock(return_value=True)
|
|
monkeypatch.setattr(
|
|
service, "_record_incident_terminal_disposition", terminal_writer
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_append_runtime_stage_receipts_to_apply", receipt_writer
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_append_alert_lifecycle_receipt", lifecycle_writer
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_verify_retry_terminal_projection_readback", verifier
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_send_controlled_apply_telegram_receipt",
|
|
telegram_sender,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_retry_telegram_receipt_acknowledged",
|
|
telegram_readback,
|
|
)
|
|
|
|
result = await service.backfill_missing_retry_terminal_projections_once()
|
|
|
|
assert result == {
|
|
"scanned": 1,
|
|
"written": 1,
|
|
"retry_receipt_written": 1,
|
|
"telegram_receipt_acknowledged": 1,
|
|
"incident_receipt_written": 1,
|
|
"lifecycle_written": 1,
|
|
"verified": 1,
|
|
"runtime_apply_executed": False,
|
|
"error": None,
|
|
}
|
|
terminal_writer.assert_awaited_once()
|
|
assert terminal_writer.await_args.kwargs["success_terminal"] is False
|
|
retry_receipt = receipt_writer.await_args_list[0].kwargs["receipts"][0]
|
|
assert retry_receipt["stage_id"] == "retry_or_rollback"
|
|
assert retry_receipt["derived_from_durable_chain"] is True
|
|
receipt = receipt_writer.await_args_list[1].kwargs["receipts"][0]
|
|
assert receipt["stage_id"] == "incident_closure"
|
|
assert receipt["derived_from_durable_chain"] is True
|
|
assert lifecycle_writer.await_args.args[1] == "EXECUTION_COMPLETED"
|
|
assert lifecycle_writer.await_args.kwargs["success"] is False
|
|
telegram_sender.assert_awaited_once()
|
|
telegram_readback.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_projection_does_not_starve_verified_success_closure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
class _RowsResult(_MappingResult):
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
super().__init__()
|
|
self._rows = rows
|
|
|
|
def all(self) -> list[dict]:
|
|
return self._rows
|
|
|
|
apply_row = {
|
|
"op_id": "00000000-0000-0000-0000-000000000104",
|
|
"status": "success",
|
|
"verifier_ready": True,
|
|
}
|
|
db = _SequenceDB(_MappingResult(), _RowsResult([apply_row]))
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
yield db
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_retry_terminal_projections_once",
|
|
AsyncMock(
|
|
return_value={
|
|
"scanned": 1,
|
|
"written": 0,
|
|
"retry_receipt_written": 0,
|
|
"telegram_receipt_acknowledged": 1,
|
|
"incident_receipt_written": 0,
|
|
"lifecycle_written": 0,
|
|
"verified": 0,
|
|
"runtime_apply_executed": False,
|
|
"error": None,
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_claim_from_apply_operation_row",
|
|
lambda _row: (_claim(), _verified_result()),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_post_apply_verifier_and_learning",
|
|
AsyncMock(
|
|
return_value={
|
|
"verification": True,
|
|
"verification_passed": True,
|
|
"verification_result": "success",
|
|
"learning": True,
|
|
"trust_learning": True,
|
|
"rag_writeback": True,
|
|
"runtime_stage_receipts": [],
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_auto_repair_execution_receipt",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_timeline_projection_receipt",
|
|
AsyncMock(return_value=None),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_runtime_stage_receipts",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
closure = AsyncMock(
|
|
return_value={
|
|
"status": "controlled_apply_closed",
|
|
"closed": True,
|
|
"telegram_receipt_acknowledged": True,
|
|
"runtime_apply_executed": False,
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_reconcile_verified_apply_closure_projections",
|
|
closure,
|
|
)
|
|
|
|
result = await service.backfill_missing_auto_repair_execution_receipts_once(
|
|
limit=1
|
|
)
|
|
|
|
assert result["retry_terminal_projection_scanned"] == 1
|
|
assert result["scanned"] == 2
|
|
assert result["incident_closure_written"] == 1
|
|
assert result["telegram_receipt_acknowledged"] == 1
|
|
closure.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broker_prioritizes_retry_terminal_projection_backfill(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_expire_stale_ansible_execution_capabilities",
|
|
AsyncMock(return_value=0),
|
|
)
|
|
receipt_backfill = AsyncMock(
|
|
return_value={
|
|
"scanned": 1,
|
|
"written": 0,
|
|
"incident_closure_written": 1,
|
|
"telegram_receipt_acknowledged": 0,
|
|
"retry_terminal_projection_scanned": 1,
|
|
"retry_terminal_projection_written": 1,
|
|
"retry_terminal_projection_verified": 1,
|
|
"retry_terminal_projection_runtime_apply_executed": False,
|
|
"error": None,
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_auto_repair_execution_receipts_once",
|
|
receipt_backfill,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"run_failed_apply_check_mode_replay_once",
|
|
AsyncMock(return_value={"scanned": 0, "replayed": 0}),
|
|
)
|
|
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_stale_pending_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_catalog_drift_failed_check_modes",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
fresh_claimer = AsyncMock(return_value=[])
|
|
monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer)
|
|
|
|
result = await service.run_pending_check_modes_once(limit=1)
|
|
|
|
assert result["claimed"] == 0
|
|
assert result["retry_terminal_projection_verified"] == 1
|
|
assert result["retry_terminal_projection_runtime_apply_executed"] is False
|
|
assert result["repair_receipt_backfill_priority_tick"] is True
|
|
receipt_backfill.assert_awaited_once()
|
|
fresh_claimer.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_worker_retry_preflight_is_query_only(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_open_failed_apply_retry_row",
|
|
AsyncMock(return_value={"op_id": "apply-op"}),
|
|
)
|
|
|
|
result = await service.preflight_failed_apply_retry_queue_once(
|
|
project_id="awoooi",
|
|
window_hours=24,
|
|
)
|
|
|
|
assert result["scanned"] == 1
|
|
assert result["replayed"] == 0
|
|
assert result["query_only"] is True
|
|
assert result["runtime_apply_executed"] is False
|
|
assert result["execution_owner"] == "awoooi-ansible-executor-broker"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("recovery_count", "expected_replayed", "expected_terminalized"),
|
|
[(0, 1, 0), (1, 0, 1)],
|
|
)
|
|
async def test_stale_pending_retry_is_reclaimed_without_runtime_apply(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
recovery_count: int,
|
|
expected_replayed: int,
|
|
expected_terminalized: int,
|
|
) -> None:
|
|
replay_op_id = "00000000-0000-0000-0000-000000000105"
|
|
apply_op_id = "00000000-0000-0000-0000-000000000104"
|
|
db = _SequenceDB(
|
|
_MappingResult(),
|
|
_MappingResult(scalar=replay_op_id),
|
|
_MappingResult(scalar=replay_op_id),
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
yield db
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_open_failed_apply_retry_row",
|
|
AsyncMock(
|
|
return_value={
|
|
"op_id": apply_op_id,
|
|
"stale_replay_op_id": replay_op_id,
|
|
"stale_replay_recovery_count": recovery_count,
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_claim_from_apply_operation_row",
|
|
lambda _row: (
|
|
_claim(),
|
|
service.AnsibleRunResult(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="failed apply",
|
|
duration_ms=10,
|
|
),
|
|
),
|
|
)
|
|
runtime_blockers = (
|
|
["ansible_binary_missing"] if expected_terminalized else []
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_runtime_blockers",
|
|
lambda: runtime_blockers,
|
|
)
|
|
transport_probe = AsyncMock(
|
|
return_value=(
|
|
["ansible_transport_unavailable"]
|
|
if expected_terminalized
|
|
else []
|
|
)
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
transport_probe,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"build_ansible_check_mode_command",
|
|
lambda **_kwargs: object(),
|
|
)
|
|
replay_runner = AsyncMock(
|
|
return_value=service.AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="check ok",
|
|
stderr="",
|
|
duration_ms=15,
|
|
)
|
|
)
|
|
monkeypatch.setattr(service, "_run_ansible_command", replay_runner)
|
|
receipt_writer = AsyncMock(return_value=True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_retry_runtime_stage_receipt",
|
|
receipt_writer,
|
|
)
|
|
|
|
result = await service.run_failed_apply_check_mode_replay_once(
|
|
timeout_seconds=30,
|
|
)
|
|
|
|
assert result["replayed"] == expected_replayed
|
|
if expected_terminalized:
|
|
assert result["check_mode_passed"] == 0
|
|
assert result["check_mode_failed"] == 1
|
|
else:
|
|
assert result["check_mode_passed"] == 1
|
|
assert result["check_mode_failed"] == 0
|
|
assert result["stale_retry_terminalized"] == expected_terminalized
|
|
assert result["runtime_apply_executed"] == 0
|
|
assert "pg_advisory_xact_lock" in db.statements[0]
|
|
assert "UPDATE automation_operation_log replay" in db.statements[1]
|
|
assert "reclaimed_after_stale_pending" in db.statements[1]
|
|
assert "input ->> 'retry_claim_id'" in db.statements[2]
|
|
persisted_output = json.loads(db.parameters[2]["output"])
|
|
persisted_dry_run = json.loads(db.parameters[2]["dry_run_result"])
|
|
assert persisted_output["check_mode_executed"] is (
|
|
not bool(expected_terminalized)
|
|
)
|
|
assert persisted_output["check_mode_replay_performed"] is (
|
|
not bool(expected_terminalized)
|
|
)
|
|
assert persisted_output["runtime_apply_executed"] is False
|
|
assert persisted_dry_run["check_mode_executed"] is (
|
|
not bool(expected_terminalized)
|
|
)
|
|
assert persisted_dry_run["check_mode_replay_performed"] is (
|
|
not bool(expected_terminalized)
|
|
)
|
|
assert persisted_dry_run["runtime_apply_executed"] is False
|
|
if expected_terminalized:
|
|
replay_runner.assert_not_awaited()
|
|
transport_probe.assert_not_awaited()
|
|
else:
|
|
replay_runner.assert_awaited_once()
|
|
transport_probe.assert_awaited_once()
|
|
receipt_writer.assert_awaited_once()
|
|
assert receipt_writer.await_args.kwargs[
|
|
"check_mode_replay_performed"
|
|
] is (not bool(expected_terminalized))
|
|
|
|
|
|
def test_candidate_worker_defaults_to_query_only_retry_preflight() -> None:
|
|
defaults = candidate_job.enqueue_missing_ansible_candidates_once.__kwdefaults__
|
|
|
|
assert defaults is not None
|
|
assert defaults["retry_replayer"] is service.preflight_failed_apply_retry_queue_once
|