awoooi/apps/api/tests/test_incident_signal_source_compatibility.py

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from pydantic import ValidationError

from src.api.v1 import auto_repair as auto_repair_api
from src.api.v1 import incidents as incidents_api
from src.api.v1 import webhooks as webhooks_api
from src.models.incident import IncidentStatus, Severity
from src.repositories import incident_repository as repository_module
from src.repositories.incident_repository import IncidentDBRepository
from src.services.incident_engine import IncidentEngineAdapter
from src.services.incident_service import IncidentService

SENSOR_SOURCES = ("journal", "node-exporter", "sensor-probe", "sensor-agent")
D037_INCIDENT_ID = "INC-20260711-D037E5"


def _record(
    incident_id: str,
    *,
    status: IncidentStatus,
    source: str | None,
    malformed: bool = False,
    auto_repair_count: int = 0,
) -> SimpleNamespace:
    timestamp = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
    signal = {
        "alert_name": f"Signal-{incident_id}",
        "annotations": {},
        "labels": {},
    }
    if not malformed:
        signal.update({
            "signal_id": incident_id[-8:],
            "severity": Severity.P2.value,
            "source": source,
            "fired_at": timestamp.isoformat(),
        })
    frequency_snapshot = None
    if auto_repair_count:
        frequency_snapshot = {
            "anomaly_key": f"anomaly-{incident_id}",
            "auto_repair_count": auto_repair_count,
            "last_repair_action": "ansible repair host service",
            "last_repair_success": True,
        }
    return SimpleNamespace(
        incident_id=incident_id,
        status=status,
        severity=Severity.P2,
        signals=[signal],
        affected_services=["awoooi-api"],
        proposal_ids=[],
        frequency_snapshot=frequency_snapshot,
        created_at=timestamp,
        updated_at=timestamp,
        resolved_at=(timestamp if status is IncidentStatus.RESOLVED else None),
        closed_at=None,
    )


class _RowsResult:
    def __init__(self, records: list[SimpleNamespace]) -> None:
        self._records = records

    def scalars(self) -> _RowsResult:
        return self

    def all(self) -> list[SimpleNamespace]:
        return self._records


class _Db:
    def __init__(self, records: list[SimpleNamespace]) -> None:
        self.records = records
        self.statements: list[object] = []

    async def execute(self, statement: object) -> _RowsResult:
        self.statements.append(statement)
        return _RowsResult(self.records)


def _install_fake_db(
    monkeypatch: pytest.MonkeyPatch,
    records: list[SimpleNamespace],
) -> tuple[_Db, list[str | None]]:
    db = _Db(records)
    project_ids: list[str | None] = []

    @asynccontextmanager
    async def fake_get_db_context(project_id: str | None = None):
        project_ids.append(project_id)
        yield db

    monkeypatch.setattr(repository_module, "get_db_context", fake_get_db_context)
    return db, project_ids


@pytest.mark.asyncio
async def test_active_readback_accepts_sensor_sources_quarantines_malformed_and_excludes_d037(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    records = [
        _record(
            f"INC-20260711-SENSOR-{index}",
            status=IncidentStatus.INVESTIGATING,
            source=source,
        )
        for index, source in enumerate(SENSOR_SOURCES, start=1)
    ]
    records.extend([
        _record(
            D037_INCIDENT_ID,
            status=IncidentStatus.RESOLVED,
            source="alertmanager",
        ),
        _record(
            "INC-20260711-3A4165",
            status=IncidentStatus.INVESTIGATING,
            source=None,
            malformed=True,
        ),
    ])
    db, project_ids = _install_fake_db(monkeypatch, records)

    readback = await IncidentDBRepository().get_active_readback(
        project_id="awoooi"
    )

    assert [incident.signals[0].source for incident in readback.incidents] == list(
        SENSOR_SOURCES
    )
    assert D037_INCIDENT_ID not in {
        incident.incident_id for incident in readback.incidents
    }
    assert [item.model_dump() for item in readback.degraded_records] == [{
        "incident_id": "INC-20260711-3A4165",
        "reason_code": "incident_signal_schema_invalid",
        "quarantined_from_response": True,
    }]
    assert project_ids == ["awoooi"]
    assert IncidentStatus.RESOLVED not in db.statements[0].compile().params["status_1"]


@pytest.mark.asyncio
async def test_unknown_source_is_fail_visible_without_blocking_valid_active_rows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    records = [
        _record(
            "INC-20260711-UNKNOWN",
            status=IncidentStatus.INVESTIGATING,
            source="unregistered-sensor",
        )
    ]
    _install_fake_db(monkeypatch, records)
    service = IncidentService()
    redis_read = AsyncMock()
    monkeypatch.setattr(service, "get_from_working_memory", redis_read)

    readback = await service.get_active_incidents_readback(project_id="awoooi")

    assert readback.incidents == []
    assert readback.degraded is True
    assert readback.degraded_records[0].reason_code == (
        "incident_signal_source_unsupported"
    )
    redis_read.assert_not_awaited()

    class _DecisionManager:
        async def _find_existing_tokens_for_incidents(self, _incident_ids):
            return {}

    monkeypatch.setattr(incidents_api, "get_incident_service", lambda: service)
    monkeypatch.setattr(
        incidents_api,
        "get_decision_manager",
        lambda: _DecisionManager(),
    )
    monkeypatch.setattr(auto_repair_api, "get_incident_service", lambda: service)

    active = await incidents_api.list_incidents(
        project_id="awoooi",
        generate_missing_decisions=False,
    )
    history = await auto_repair_api.get_repair_history(
        project_id="awoooi",
        limit=20,
    )

    for response in (active, history):
        assert response.degraded is True
        assert response.degraded_count == 1
        assert response.degraded_records[0].incident_id == (
            "INC-20260711-UNKNOWN"
        )
        assert response.redis_fallback_used is False


@pytest.mark.asyncio
async def test_auto_repair_history_remains_usable_with_journal_and_malformed_rows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    records = [
        _record(
            "INC-20260711-JOURNAL-HISTORY",
            status=IncidentStatus.INVESTIGATING,
            source="journal",
            auto_repair_count=1,
        ),
        _record(
            D037_INCIDENT_ID,
            status=IncidentStatus.RESOLVED,
            source="alertmanager",
            auto_repair_count=1,
        ),
        _record(
            "INC-20260711-3A4165",
            status=IncidentStatus.INVESTIGATING,
            source=None,
            malformed=True,
        ),
    ]
    _install_fake_db(monkeypatch, records)
    monkeypatch.setattr(
        auto_repair_api,
        "get_incident_service",
        IncidentService,
    )

    history = await auto_repair_api.get_repair_history(
        project_id="awoooi",
        limit=20,
    )

    assert history.count == 1
    assert history.items[0].incident_id == "INC-20260711-JOURNAL-HISTORY"
    assert D037_INCIDENT_ID not in {item.incident_id for item in history.items}
    assert history.degraded is True
    assert history.degraded_records[0].incident_id == "INC-20260711-3A4165"
    assert history.degraded_records[0].reason_code == (
        "incident_signal_schema_invalid"
    )


def test_signal_ingress_rejects_unregistered_source_before_stream_write() -> None:
    with pytest.raises(ValidationError):
        webhooks_api.SignalPayload(
            source="unregistered-sensor",
            alert_name="UnknownProducer",
            severity="warning",
            namespace="infra",
            target="host",
        )


@pytest.mark.asyncio
async def test_incident_engine_rejects_unknown_source_before_durable_processing() -> None:
    brain_engine = SimpleNamespace(process_signal=AsyncMock(return_value=None))
    adapter = IncidentEngineAdapter(brain_engine)

    assert await adapter.process_signal({"source": "journal"}) is None
    brain_engine.process_signal.assert_awaited_once_with({"source": "journal"})

    with pytest.raises(ValueError, match="unsupported_signal_source"):
        await adapter.process_signal({"source": "unregistered-sensor"})
    assert brain_engine.process_signal.await_count == 1