awoooi/apps/api/tests/test_repair_candidate_service.py

from datetime import datetime, timezone
from uuid import uuid4

import pytest

from src.models.approval import ApprovalRequestCreate, BlastRadius, DataImpact, RiskLevel
from src.models.incident import Incident, Severity, Signal
from src.models.playbook import (
    ActionType,
    Playbook,
    PlaybookSource,
    PlaybookStatus,
    RepairStep,
)
from src.models.playbook import RiskLevel as PlaybookRiskLevel
from src.services.approval_db import approval_request_to_record_data
from src.services.evidence_snapshot import EvidenceSnapshot
from src.services.repair_candidate_service import RepairCandidateService


class FakeInvestigator:
    def __init__(self, evidence: EvidenceSnapshot | None) -> None:
        self.evidence = evidence

    async def investigate(self, incident: Incident) -> EvidenceSnapshot | None:
        return self.evidence


class FakePlaybookRepository:
    def __init__(self, playbook: Playbook | None) -> None:
        self.playbook = playbook

    async def get_by_id(self, playbook_id: str) -> Playbook | None:
        if self.playbook and self.playbook.playbook_id == playbook_id:
            return self.playbook
        return None


class FakeIncidentService:
    async def get_from_working_memory(self, incident_id: str) -> None:
        return None


class FakeAutoRepairService:
    def preview_write_ssh_mcp_route(self, incident: Incident, command: str) -> bool:
        return True


def _incident() -> Incident:
    return Incident(
        incident_id="INC-TEST-REPAIR",
        severity=Severity.P2,
        signals=[
            Signal(
                alert_name="NodeExporterDown",
                severity=Severity.P2,
                source="alertmanager",
                fired_at=datetime.now(timezone.utc),
                labels={"namespace": "awoooi-prod", "deployment": "awoooi-api"},
                annotations={"summary": "node exporter down"},
            )
        ],
        affected_services=["awoooi-api"],
    )


def _evidence(incident_id: str, *, sensors_succeeded: int = 2) -> EvidenceSnapshot:
    return EvidenceSnapshot(
        incident_id=incident_id,
        sensors_attempted=3,
        sensors_succeeded=sensors_succeeded,
        mcp_health={"k8s": sensors_succeeded > 0, "prometheus": sensors_succeeded > 1},
        evidence_summary="k8s events and metrics collected",
    )


def _playbook(command: str, *, risk_level: PlaybookRiskLevel = PlaybookRiskLevel.LOW) -> Playbook:
    return Playbook(
        playbook_id="PB-REPAIR-001",
        name="重啟 API deployment",
        description="以 approved PlayBook 修復 API 工作負載",
        status=PlaybookStatus.APPROVED,
        source=PlaybookSource.MANUAL,
        trust_score=0.72,
        estimated_duration_minutes=4,
        repair_steps=[
            RepairStep(
                step_number=1,
                action_type=ActionType.KUBECTL,
                command=command,
                expected_result="deployment restarted and rollout verified",
                risk_level=risk_level,
                requires_approval=True,
            )
        ],
    )


def _generic_fallback_playbook() -> Playbook:
    playbook = _playbook(
        "kubectl rollout restart deployment/{target} -n {namespace}",
        risk_level=PlaybookRiskLevel.MEDIUM,
    )
    playbook.playbook_id = "PB-GENERIC-FALLBACK"
    playbook.name = "通用兜底規則"
    playbook.symptom_pattern.alert_names = ["*"]
    return playbook


@pytest.mark.asyncio
async def test_build_candidate_from_mcp_evidence_and_approved_playbook() -> None:
    incident = _incident()
    service = RepairCandidateService(
        incident_service=FakeIncidentService(),
        investigator=FakeInvestigator(_evidence(incident.incident_id)),
        playbook_repository=FakePlaybookRepository(
            _playbook("kubectl rollout restart deployment/awoooi-api -n awoooi-prod")
        ),
        auto_repair_service=FakeAutoRepairService(),
    )

    result = await service.build_from_incident(
        incident=incident,
        alertname="NodeExporterDown",
        target_resource="awoooi-api",
        namespace="awoooi-prod",
        message="node exporter is down",
        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
        matched_playbook_id="PB-REPAIR-001",
        severity="medium",
    )

    assert result.candidate_found is True
    assert result.approval_request is not None
    assert result.approval_request.action == "kubectl rollout restart deployment/awoooi-api -n awoooi-prod"
    assert result.approval_request.risk_level == RiskLevel.MEDIUM
    assert result.approval_request.matched_playbook_id == "PB-REPAIR-001"
    assert result.metadata["repair_candidate_status"] == "candidate_ready_for_approval"
    assert result.metadata["mcp_evidence"]["sensors_succeeded"] == 2
    assert result.metadata["playbook_trust"]["trust_score"] == 0.72
    assert result.metadata["verifier_plan"]


@pytest.mark.asyncio
async def test_candidate_blocked_when_mcp_evidence_missing() -> None:
    incident = _incident()
    service = RepairCandidateService(
        incident_service=FakeIncidentService(),
        investigator=FakeInvestigator(_evidence(incident.incident_id, sensors_succeeded=0)),
        playbook_repository=FakePlaybookRepository(
            _playbook("kubectl rollout restart deployment/awoooi-api -n awoooi-prod")
        ),
        auto_repair_service=FakeAutoRepairService(),
    )

    result = await service.build_from_incident(
        incident=incident,
        alertname="NodeExporterDown",
        target_resource="awoooi-api",
        namespace="awoooi-prod",
        message="node exporter is down",
        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
        matched_playbook_id="PB-REPAIR-001",
        severity="medium",
    )

    assert result.candidate_found is False
    assert "mcp_evidence_missing" in result.blockers
    assert result.metadata["repair_candidate_status"] == "blocked"


@pytest.mark.asyncio
async def test_candidate_blocked_when_playbook_is_observe_only() -> None:
    incident = _incident()
    service = RepairCandidateService(
        incident_service=FakeIncidentService(),
        investigator=FakeInvestigator(_evidence(incident.incident_id)),
        playbook_repository=FakePlaybookRepository(
            _playbook("kubectl get pods -n awoooi-prod")
        ),
        auto_repair_service=FakeAutoRepairService(),
    )

    result = await service.build_from_incident(
        incident=incident,
        alertname="NodeExporterDown",
        target_resource="awoooi-api",
        namespace="awoooi-prod",
        message="node exporter is down",
        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
        matched_playbook_id="PB-REPAIR-001",
        severity="medium",
    )

    assert result.candidate_found is False
    assert "playbook_observe_only" in result.blockers


@pytest.mark.asyncio
async def test_candidate_blocked_when_playbook_is_generic_fallback() -> None:
    incident = _incident()
    service = RepairCandidateService(
        incident_service=FakeIncidentService(),
        investigator=FakeInvestigator(_evidence(incident.incident_id)),
        playbook_repository=FakePlaybookRepository(_generic_fallback_playbook()),
        auto_repair_service=FakeAutoRepairService(),
    )

    result = await service.build_from_incident(
        incident=incident,
        alertname="UnknownAlert",
        target_resource="awoooi-api",
        namespace="awoooi-prod",
        message="unknown alert",
        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
        matched_playbook_id="PB-GENERIC-FALLBACK",
        severity="medium",
    )

    assert result.candidate_found is False
    assert "playbook_generic_fallback_not_repair" in result.blockers
    assert "通用兜底" in result.metadata["repair_candidate_blocker_summary"]


def test_approval_record_data_uses_preallocated_id_without_leaking_metadata() -> None:
    approval_id = str(uuid4())
    request = ApprovalRequestCreate(
        action="kubectl rollout restart deployment/awoooi-api -n awoooi-prod",
        description="candidate",
        risk_level=RiskLevel.MEDIUM,
        requested_by="repair-candidate-test",
        blast_radius=BlastRadius(data_impact=DataImpact.WRITE),
        metadata={"preallocated_approval_id": approval_id, "source": "test"},
    )

    data = approval_request_to_record_data(request, RiskLevel.MEDIUM, required_sigs=1)

    assert data["id"] == approval_id
    assert data["extra_metadata"] == {"source": "test"}