f453359e34
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m21s
CD Pipeline / build-and-deploy (push) Successful in 6m53s
CD Pipeline / post-deploy-checks (push) Successful in 1m55s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 0s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 49s
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
"""Alertmanager high-confidence routing remains guarded and queue-only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.api.v1 import webhooks
|
|
from src.models.approval import RiskLevel
|
|
|
|
|
|
def _analysis(
|
|
*,
|
|
confidence: float = 0.90,
|
|
action_title: str = "restart pod",
|
|
kubectl: str = "kubectl rollout restart deployment/foo",
|
|
) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
confidence=confidence,
|
|
action_title=action_title,
|
|
kubectl_command=kubectl,
|
|
)
|
|
|
|
|
|
def _can_route(analysis: SimpleNamespace, risk_level: RiskLevel) -> bool:
|
|
from src.services.action_parser import is_safe_kubectl_action
|
|
|
|
command = str(analysis.kubectl_command or "").strip()
|
|
return (
|
|
bool(command)
|
|
and analysis.confidence >= 0.85
|
|
and risk_level != RiskLevel.CRITICAL
|
|
and "NO_ACTION" not in str(analysis.action_title or "")
|
|
and is_safe_kubectl_action(command)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("analysis", "risk_level", "expected"),
|
|
[
|
|
(_analysis(), RiskLevel.MEDIUM, True),
|
|
(_analysis(confidence=0.70), RiskLevel.MEDIUM, False),
|
|
(_analysis(), RiskLevel.CRITICAL, False),
|
|
(_analysis(kubectl=""), RiskLevel.MEDIUM, False),
|
|
(_analysis(action_title="NO_ACTION: no fix"), RiskLevel.MEDIUM, False),
|
|
(
|
|
_analysis(kubectl="kubectl delete pods --all -n prod"),
|
|
RiskLevel.MEDIUM,
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_cs3_controlled_queue_eligibility(
|
|
analysis: SimpleNamespace,
|
|
risk_level: RiskLevel,
|
|
expected: bool,
|
|
) -> None:
|
|
assert _can_route(analysis, risk_level) is expected
|
|
|
|
|
|
def test_alertmanager_source_uses_queue_not_approval_executor() -> None:
|
|
source = inspect.getsource(webhooks._process_new_alert_background)
|
|
|
|
assert "_try_auto_repair_background" in source
|
|
assert "execute_approved_action" not in source
|
|
assert "ApprovalExecutionService" not in source
|
|
assert "execute_auto_repair(" not in source
|