from __future__ import annotations from collections.abc import Mapping from unittest.mock import AsyncMock import pytest from src.jobs.kubernetes_controlled_candidate_job import ( run_kubernetes_controlled_candidate_once, ) from src.services.executor import ExecutionResult, OperationType from src.services.kubernetes_controlled_candidate_service import ( enqueue_ai_decision_kubernetes_candidate, run_claimed_kubernetes_controlled_candidate, ) class _Incident: project_id = "awoooi" incident_id = "INC-20260711-D037E5" def __init__(self, target: str = "awoooi-auto-repair-canary") -> None: self.target = target def model_dump(self, *, mode: str = "json") -> dict[str, object]: assert mode == "json" return { "incident_id": self.incident_id, "project_id": self.project_id, "affected_services": [self.target], "signals": [ { "alert_name": "AwoooPAutoRepairCanaryT16", "labels": { "alertname": "AwoooPAutoRepairCanaryT16", "namespace": "awoooi-prod", "deployment": self.target, "host_type": "kubernetes", }, } ], } def _proposal() -> dict[str, str]: return { "kubectl_command": ( "kubectl rollout restart " "deployment/awoooi-auto-repair-canary -n awoooi-prod" ), "risk_level": "medium", "matched_playbook_id": "pb:auto-repair-canary", } @pytest.mark.asyncio async def test_d037_queues_typed_kubernetes_candidate_without_ansible_fallback() -> ( None ): persisted: list[dict[str, object]] = [] async def persist(record: dict[str, object], project_id: str) -> dict[str, object]: assert project_id == "awoooi" persisted.append(record) return { "event_id": "00000000-0000-0000-0000-000000000033", "created": True, "record": record, } handoff = await enqueue_ai_decision_kubernetes_candidate( incident=_Incident(), proposal_data=_proposal(), persist_candidate=persist, ) assert handoff["status"] == "controlled_kubernetes_candidate_queued" assert handoff["queued"] is True assert handoff["single_writer_executor"] == ( "awoooi-kubernetes-executor-broker" ) assert handoff["typed_domain"] == "kubernetes_workload" assert handoff["canonical_asset_id"] == ("service:awoooi-auto-repair-canary") assert handoff["cross_domain_fallback_allowed"] is False assert len(persisted) == 1 record = persisted[0] assert record["legacy_ansible_candidate_consulted"] is False assert record["runtime_mutation_performed"] is False claim = record["claim"] assert isinstance(claim, Mapping) assert claim["executor"] == "kubernetes_controlled_executor" assert claim["verifier"] == "kubernetes_rollout_verifier" @pytest.mark.asyncio async def test_exact_duplicate_reuses_durable_candidate_instead_of_rebuilding() -> None: first_record: dict[str, object] = {} async def persist(record: dict[str, object], _project_id: str) -> dict[str, object]: nonlocal first_record if not first_record: first_record = record created = True else: created = False return { "event_id": "00000000-0000-0000-0000-000000000033", "created": created, "record": first_record, } first = await enqueue_ai_decision_kubernetes_candidate( incident=_Incident(), proposal_data=_proposal(), persist_candidate=persist, ) duplicate = await enqueue_ai_decision_kubernetes_candidate( incident=_Incident(), proposal_data=_proposal(), persist_candidate=persist, ) assert first["deduplicated"] is False assert duplicate["deduplicated"] is True assert duplicate["status"] == "controlled_kubernetes_candidate_exists" assert duplicate["claim_id"] == first["claim_id"] assert duplicate["automation_run_id"] == first["automation_run_id"] @pytest.mark.asyncio async def test_unknown_asset_fails_closed_without_candidate_or_runtime_mutation() -> ( None ): persist = AsyncMock() handoff = await enqueue_ai_decision_kubernetes_candidate( incident=_Incident(target="unregistered-canary"), proposal_data={ "kubectl_command": ( "kubectl rollout restart deployment/unregistered-canary " "-n awoooi-prod" ) }, persist_candidate=persist, ) assert handoff["status"] == "kubernetes_candidate_blocked_no_write" assert handoff["queued"] is False assert handoff["runtime_mutation_performed"] is False assert handoff["cross_domain_fallback_allowed"] is False assert handoff["active_blockers"] persist.assert_not_awaited() def _claimed_candidate() -> dict[str, object]: return { "schema_version": "kubernetes_controlled_candidate_v1", "status": "claimed", "event_id": "00000000-0000-0000-0000-000000000033", "claim_token": "00000000-0000-0000-0000-000000000034", "automation_run_id": "00000000-0000-0000-0000-000000000035", "matched_playbook_id": "pb:auto-repair-canary", "claim": { "claim_id": "k8s-claim:d037e5", "incident_id": "INC-20260711-D037E5", "canonical_asset_id": "service:awoooi-auto-repair-canary", "workload_kind": "deployment", "workload_name": "awoooi-auto-repair-canary", "namespace": "awoooi-prod", }, } def _verified_execution(run_id: str = "run:k8s:d037e5") -> ExecutionResult: return ExecutionResult( success=True, message="verified", operation_type=OperationType.RESTART_DEPLOYMENT, target_resource="deployment/awoooi-auto-repair-canary", namespace="awoooi-prod", duration_ms=25, k8s_response={ "runtime_write_performed": True, "runtime_write_outcome": "confirmed_applied", "lifecycle_receipt": { "receipt_id": "lifecycle:d037e5", "run_id": run_id, "durable_writeback_ack": True, }, "verifier_receipt": { "receipt_id": "verifier:d037e5", "run_id": run_id, "verified": True, "durable_writeback_ack": True, }, }, ) @pytest.mark.asyncio async def test_single_dispatch_closes_only_with_same_run_receipts() -> None: dispatch_started = False execute = AsyncMock(return_value=_verified_execution()) finalized: list[Mapping[str, object]] = [] async def begin(_candidate: Mapping[str, object], _project: str) -> bool: nonlocal dispatch_started if dispatch_started: return False dispatch_started = True return True async def learning( _candidate: Mapping[str, object], _execution: object, run_id: str, ) -> dict[str, object]: return { "receipt_id": "learning:d037e5", "run_id": run_id, "durable_writeback_ack": True, } async def telegram( _candidate: Mapping[str, object], _execution: object, run_id: str, ) -> dict[str, object]: return { "receipt_id": "telegram:d037e5", "run_id": run_id, "provider_delivery_acknowledged": True, "durable_writeback_ack": True, } async def finalize( _candidate: Mapping[str, object], _project: str, closure: Mapping[str, object], ) -> bool: finalized.append(closure) return True first = await run_claimed_kubernetes_controlled_candidate( _claimed_candidate(), begin_dispatch=begin, execute_claim=execute, write_learning_receipt=learning, send_telegram_receipt=telegram, finalize_candidate=finalize, ) duplicate = await run_claimed_kubernetes_controlled_candidate( _claimed_candidate(), begin_dispatch=begin, execute_claim=execute, write_learning_receipt=learning, send_telegram_receipt=telegram, finalize_candidate=finalize, ) assert first["status"] == "verified_closed" assert first["closed"] is True assert first["same_run_receipts"] is True assert first["durable_closure_ack"] is True assert first["replay_apply_allowed"] is False assert duplicate["status"] == "duplicate_apply_suppressed" assert duplicate["runtime_mutation_performed"] is False execute.assert_awaited_once() assert len(finalized) == 1 @pytest.mark.asyncio async def test_run_mismatch_is_durable_pending_closure_and_never_reapply() -> None: execution = _verified_execution() execute = AsyncMock(return_value=execution) finalized: list[Mapping[str, object]] = [] async def begin(_candidate: Mapping[str, object], _project: str) -> bool: return True async def learning( _candidate: Mapping[str, object], _execution: object, _run_id: str, ) -> dict[str, object]: return { "run_id": "run:k8s:wrong", "durable_writeback_ack": True, } async def telegram( _candidate: Mapping[str, object], _execution: object, run_id: str, ) -> dict[str, object]: return { "run_id": run_id, "provider_delivery_acknowledged": True, "durable_writeback_ack": True, } async def finalize( _candidate: Mapping[str, object], _project: str, closure: Mapping[str, object], ) -> bool: finalized.append(closure) return True result = await run_claimed_kubernetes_controlled_candidate( _claimed_candidate(), begin_dispatch=begin, execute_claim=execute, write_learning_receipt=learning, send_telegram_receipt=telegram, finalize_candidate=finalize, ) assert result["status"] == "applied_pending_closure" assert result["closed"] is False assert result["same_run_receipts"] is False assert result["replay_apply_allowed"] is False assert finalized[0]["runtime_mutation_performed"] is True @pytest.mark.asyncio async def test_dedicated_worker_claims_only_one_candidate_per_tick() -> None: claim_next = AsyncMock(return_value=_claimed_candidate()) run_candidate = AsyncMock( return_value={"status": "verified_closed", "closed": True} ) stats = await run_kubernetes_controlled_candidate_once( batch_limit=99, claim_next=claim_next, run_candidate=run_candidate, ) assert stats["claimed"] == 1 assert stats["verified_closed"] == 1 assert stats["single_writer_executor"] == ( "awoooi-kubernetes-executor-broker" ) claim_next.assert_awaited_once_with(project_id="awoooi") run_candidate.assert_awaited_once()