f19b608005
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m39s
CD Pipeline / build-and-deploy (push) Successful in 6m25s
CD Pipeline / post-deploy-checks (push) Successful in 2m9s
428 lines
13 KiB
Python
428 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job
|
|
from src.services import awooop_ansible_check_mode_service as service
|
|
|
|
|
|
def _claim() -> service.AnsibleCheckModeClaim:
|
|
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",
|
|
},
|
|
)
|
|
|
|
|
|
def _verified_result() -> service.AnsibleRunResult:
|
|
return service.AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="",
|
|
duration_ms=25,
|
|
post_verifier_passed=True,
|
|
)
|
|
|
|
|
|
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_set(" in source
|
|
assert "'legacy_outcome'" in source
|
|
assert "'{automation_terminal}'" in source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
terminal_writer = AsyncMock()
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_read_verified_apply_closure_prerequisites",
|
|
AsyncMock(
|
|
return_value={
|
|
"ready": False,
|
|
"missing": ["telegram_receipt"],
|
|
"receipts": {},
|
|
}
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_incident_terminal_disposition",
|
|
terminal_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"],
|
|
}
|
|
terminal_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}}),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_incident_terminal_disposition",
|
|
AsyncMock(
|
|
return_value={
|
|
"automation_run_id": (
|
|
"00000000-0000-0000-0000-000000000102"
|
|
),
|
|
"apply_op_id": "00000000-0000-0000-0000-000000000104",
|
|
"incident_resolved": True,
|
|
}
|
|
),
|
|
)
|
|
receipt_writer = AsyncMock(return_value=True)
|
|
lifecycle_writer = AsyncMock(return_value=True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_append_runtime_stage_receipts_to_apply",
|
|
receipt_writer,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_append_alert_lifecycle_receipt",
|
|
lifecycle_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
|
|
closure_receipt = receipt_writer.await_args.kwargs["receipts"][0]
|
|
assert closure_receipt["stage_id"] == "incident_closure"
|
|
assert closure_receipt["durable_receipt"] is True
|
|
assert lifecycle_writer.await_args.args[1] == "RESOLVED"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_projection_replay_sends_missing_receipt_without_reapplying(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
readback = AsyncMock(
|
|
side_effect=[
|
|
{"receipts": {"telegram_receipt": False}},
|
|
{"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()
|
|
|
|
|
|
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 = "\n".join(
|
|
str(value)
|
|
for value in candidate_job._fetch_missing_candidate_incidents.__code__.co_consts
|
|
)
|
|
|
|
assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103"
|
|
assert "approval_records" in query_source
|
|
assert "ansible_execution_skipped" in query_source
|
|
assert "terminal.parent_op_id = existing.op_id" in query_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broker_reconciles_receipts_before_claiming_fresh_work(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
retry_replayer = AsyncMock()
|
|
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,
|
|
)
|
|
|
|
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_not_awaited()
|
|
|
|
|
|
@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",
|
|
"terminal_type": "no_write_replay_passed_waiting_repair_candidate",
|
|
"telegram_receipt_acknowledged": True,
|
|
}
|
|
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)
|
|
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
|
|
)
|
|
|
|
result = await service.backfill_missing_retry_terminal_projections_once()
|
|
|
|
assert result == {
|
|
"scanned": 1,
|
|
"written": 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
|
|
receipt = receipt_writer.await_args.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
|
|
|
|
|
|
@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,
|
|
)
|
|
|
|
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()
|
|
|
|
|
|
@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"
|
|
|
|
|
|
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
|