wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(agent): keep closure lanes moving safely

Browse Source
This commit is contained in:
ogt
2026-07-11 16:09:31 +08:00
parent c4a77a1e51
commit bd27d4e759
6 changed files with 1675 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

443
apps/api/tests/test_ansible_verified_closure.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import inspect
from contextlib import asynccontextmanager
from unittest.mock import AsyncMock
import pytest
@@ -9,6 +10,36 @@ from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job
from src.services import awooop_ansible_check_mode_service as service
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()
def _claim() -> service.AnsibleCheckModeClaim:
return service.AnsibleCheckModeClaim(
op_id="00000000-0000-0000-0000-000000000101",
@@ -61,25 +92,69 @@ 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 "jsonb_build_object(" in source
assert "'legacy_outcome'" in source
assert "'{automation_terminal}'" in source
def test_incident_terminal_persists_automated_outcome_truth() -> None:
source = inspect.getsource(service._record_incident_terminal_disposition)
@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,
},
}
),
)
assert "'proposal_executed'" in source
assert "'execution_success'" in source
assert 'outcome.get("proposal_executed") is True' in source
assert 'outcome.get("execution_success") is True' in source
@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:
terminal_writer = AsyncMock()
atomic_writer = AsyncMock()
monkeypatch.setattr(
service,
"_read_verified_apply_closure_prerequisites",
@@ -93,8 +168,8 @@ async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
)
monkeypatch.setattr(
service,
"_record_incident_terminal_disposition",
terminal_writer,
"_commit_verified_incident_closure",
atomic_writer,
)
result = await service._finalize_verified_apply_closure(
@@ -108,7 +183,7 @@ async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
"closed": False,
"missing": ["telegram_receipt"],
}
terminal_writer.assert_not_awaited()
atomic_writer.assert_not_awaited()
@pytest.mark.asyncio
@@ -120,30 +195,19 @@ async def test_closure_resolves_only_after_durable_readback(
"_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,
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,
}
)
monkeypatch.setattr(
service,
"_append_alert_lifecycle_receipt",
lifecycle_writer,
"_commit_verified_incident_closure",
atomic_writer,
)
monkeypatch.setattr(
service,
@@ -165,10 +229,65 @@ async def test_closure_resolves_only_after_durable_readback(
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"
atomic_writer.assert_awaited_once()
assert atomic_writer.await_args.kwargs["closure_prerequisite_count"] == 1
@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
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 "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 latest.created_at DESC, latest.op_id DESC" in (
recorder_source
)
@pytest.mark.asyncio
@@ -234,15 +353,35 @@ def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -
"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
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 "ansible_execution_skipped" in query_source
assert "terminal.parent_op_id = existing.op_id" in query_source
assert "verified_ansible_candidate_terminal_sql" in query_source
assert "AND NOT ({terminal_candidate_sql})" in query_source
assert "ORDER BY latest.created_at DESC, latest.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
@@ -326,6 +465,72 @@ async def test_broker_reconciles_receipts_then_continues_claiming_fresh_work(
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,
@@ -333,8 +538,13 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
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": True,
"telegram_receipt_acknowledged": False,
}
monkeypatch.setattr(
service,
@@ -366,6 +576,8 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
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
)
@@ -378,12 +590,24 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
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,
@@ -392,11 +616,16 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
}
terminal_writer.assert_awaited_once()
assert terminal_writer.await_args.kwargs["success_terminal"] is False
receipt = receipt_writer.await_args.kwargs["receipts"][0]
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
@@ -426,6 +655,29 @@ async def test_broker_prioritizes_retry_terminal_projection_backfill(
"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)
@@ -434,6 +686,7 @@ async def test_broker_prioritizes_retry_terminal_projection_backfill(
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
@@ -458,6 +711,108 @@ async def test_signal_worker_retry_preflight_is_query_only(
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,
),
),
)
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
monkeypatch.setattr(
service,
"recent_ansible_transport_blockers",
AsyncMock(return_value=[]),
)
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]
if expected_terminalized:
replay_runner.assert_not_awaited()
else:
replay_runner.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__

18
apps/api/tests/test_awooop_ansible_candidate_backfill_job.py
View File

@@ -130,7 +130,7 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo
@pytest.mark.asyncio
async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candidates(
async def test_backfill_runs_retry_then_continues_fresh_candidates(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
@@ -147,11 +147,10 @@ async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candida
"error": None,
}
async def fresh_scan_must_not_run(**_kwargs):
raise AssertionError("fresh candidates must wait for the open retry tick")
fresh_scan = AsyncMock(return_value=[])
monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True)
monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run)
monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan)
receipt_backfiller = AsyncMock(return_value={"written": 0, "error": None})
recorder = AsyncMock(return_value=True)
@@ -167,12 +166,13 @@ async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candida
assert result["failed_apply_retry_stage_receipt_written"] == 1
assert result["failed_apply_retry_priority_tick"] is True
assert result["queued"] == 0
receipt_backfiller.assert_not_awaited()
receipt_backfiller.assert_awaited_once()
fresh_scan.assert_awaited_once()
recorder.assert_not_awaited()
@pytest.mark.asyncio
async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work(
async def test_backfill_surfaces_retry_blockers_and_continues_fresh_scan(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
@@ -185,11 +185,10 @@ async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work(
"error": None,
}
async def fresh_scan_must_not_run(**_kwargs):
raise AssertionError("fresh candidates must not grow behind retry blockers")
fresh_scan = AsyncMock(return_value=[])
monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True)
monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run)
monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan)
result = await job.enqueue_missing_ansible_candidates_once(
retry_replayer=fake_retry_replayer,
@@ -201,6 +200,7 @@ async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work(
]
assert result["failed_apply_retry_priority_tick"] is True
assert result["queued"] == 0
fresh_scan.assert_awaited_once()
@pytest.mark.asyncio

42
apps/api/tests/test_awooop_truth_chain_service.py
View File

@@ -17,6 +17,7 @@ from src.services.awooop_ansible_audit_service import (
build_ansible_decision_audit_payload,
build_ansible_truth,
record_ansible_decision_audit,
verified_ansible_candidate_terminal_sql,
)
from src.services.awooop_ansible_check_mode_service import (
_EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE,
@@ -26,6 +27,7 @@ from src.services.awooop_ansible_check_mode_service import (
_append_runtime_stage_receipts_to_apply,
_automation_operation_log_incident_id,
_build_auto_repair_execution_receipt,
_build_retry_runtime_stage_receipt,
_claim_from_apply_operation_row,
_claim_from_stale_check_mode_row,
_execution_capability_timeout_seconds,
@@ -147,13 +149,14 @@ def test_quality_summary_uses_batched_truth_chain_inputs() -> None:
def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> None:
decision_source = inspect.getsource(record_ansible_decision_audit)
terminal_source = verified_ansible_candidate_terminal_sql("candidate")
claim_source = inspect.getsource(claim_pending_check_modes)
assert "operation_type, actor, status, incident_id" in decision_source
assert "candidate.incident_id::text" in decision_source
assert "candidate.input ->> 'incident_id'" in decision_source
assert "terminal.parent_op_id = candidate.op_id" in decision_source
assert "ansible_execution_skipped" in decision_source
assert "verified_ansible_candidate_terminal_sql" in decision_source
assert "ansible_execution_skipped" in terminal_source
assert "operation_type, actor, status, incident_id" in claim_source
assert "incident_db_id" in decision_source
assert "incident_db_id" in claim_source
@@ -2108,7 +2111,7 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims
@pytest.mark.asyncio
async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims(
async def test_execution_broker_runs_retry_then_keeps_fresh_capacity_moving(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.services import awooop_ansible_check_mode_service as service
@@ -2132,7 +2135,7 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla
"error": None,
}
async def candidate_claim_must_not_run(**_kwargs):
async def candidate_claim_continues(**_kwargs):
nonlocal claim_attempted
claim_attempted = True
return []
@@ -2161,22 +2164,22 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla
monkeypatch.setattr(
service,
"claim_stale_pending_check_modes",
candidate_claim_must_not_run,
candidate_claim_continues,
)
monkeypatch.setattr(
service,
"claim_catalog_drift_failed_check_modes",
candidate_claim_must_not_run,
candidate_claim_continues,
)
monkeypatch.setattr(
service,
"claim_pending_check_modes",
candidate_claim_must_not_run,
candidate_claim_continues,
)
result = await service.run_pending_check_modes_once(limit=1)
assert claim_attempted is False
assert claim_attempted is True
assert result["claimed"] == 0
assert result["failed_apply_retry_scanned"] == 1
assert result["failed_apply_retry_replayed"] == 1
@@ -2424,10 +2427,11 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
source = inspect.getsource(run_failed_apply_check_mode_replay_once)
preflight_source = inspect.getsource(_load_open_failed_apply_retry_row)
assert "FOR UPDATE SKIP LOCKED" in preflight_source
assert "FOR UPDATE OF apply SKIP LOCKED" in preflight_source
assert "controlled_retry_check_mode_replay" in source
assert "build_ansible_check_mode_command" in source
assert "verifier.incident_id = coalesce(" in preflight_source
assert "verification_result IN ('failed', 'timeout')" in preflight_source
assert "apply.input ->> 'incident_id'" in preflight_source
assert "controlled_apply_allowed=True" in source
assert '"approval_required_before_apply": False' in source
@@ -2436,16 +2440,26 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
assert "queue_ai_playbook_or_transport_repair_candidate" in source
assert "retry_requires_repair_and_new_apply_gate" not in source
assert "_record_retry_runtime_stage_receipt" in source
assert "NOT EXISTS" in preflight_source
assert "failed_apply_retry_already_claimed" in source
assert "LEFT JOIN LATERAL" in preflight_source
assert "replay.op_id IS NULL" in preflight_source
assert "replay.status = 'pending'" in preflight_source
assert "failed_apply_retry_claim_raced" in source
assert "reclaimed_after_stale_pending" in source
assert "retry_claim_id" in source
assert "retry_recovery_count" in source
assert "pg_advisory_xact_lock" in source
assert "run_controlled_apply_for_claim" not in source
receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt)
receipt_builder_source = inspect.getsource(
_build_retry_runtime_stage_receipt
)
append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply)
assert "_runtime_stage_receipt" in receipt_source
assert 'stage_id="retry_or_rollback"' in receipt_source
assert "_build_retry_runtime_stage_receipt" in receipt_source
assert "_runtime_stage_receipt" in receipt_builder_source
assert 'stage_id="retry_or_rollback"' in receipt_builder_source
assert "jsonb_array_elements" in append_source
assert "ansible_controlled_retry_terminal_v2" in receipt_source
assert "ansible_controlled_retry_terminal_v2" in receipt_builder_source
assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source
assert "runtime_apply_executed" in receipt_source