346 lines
11 KiB
Python
346 lines
11 KiB
Python
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()
|