awoooi/apps/api/tests/test_agent99_sre_bridge.py

from __future__ import annotations

import json

import pytest

from src.services.agent99_sre_bridge import (
    agent99_sre_single_flight_key,
    bridge_alertmanager_to_agent99,
    build_agent99_sre_alert,
    post_agent99_sre_alert,
    release_agent99_sre_single_flight,
    resolve_agent99_alert_kind,
    write_agent99_sre_alert,
)


def test_provider_freshness_alert_builds_agent99_contract() -> None:
    payload = build_agent99_sre_alert(
        alert_id="alert-20260709233000",
        alertname="provider_freshness_signal",
        severity="warning",
        namespace="awoooi-prod",
        target_resource="provider_freshness_signal",
        message=(
            "raw_signal_normalized controlled_runtime_candidate freshness "
            "last seen missing direct ref verifier readback"
        ),
        labels={
            "service": "awoooi",
            "token": "must-not-leak",
        },
        annotations={
            "summary": "Provider freshness evidence is missing",
        },
        fingerprint="fp-provider-freshness",
        alert_category="source_provider_freshness_triage",
        notification_type="raw_signal_normalized",
        source_url="https://awoooi.wooo.work/api/v1/webhooks/alertmanager",
    )

    assert payload["source"] == "awoooi-api-alertmanager"
    assert payload["kind"] == "provider_freshness_signal"
    assert payload["suggestedMode"] == "ProviderFreshness"
    assert payload["controlledApply"] is False
    assert payload["force"] is True
    assert "provider-freshness" in payload["labels"]
    assert "no-false-green" in payload["labels"]
    assert "token=[redacted]" in payload["labels"]
    assert "must-not-leak" not in json.dumps(payload)
    assert "do not switch provider" in payload["instruction"]
    assert payload["routing"]["routeSource"] == "structured_kind"
    assert payload["routing"]["correlationKey"] == payload["routing"]["groupKey"]


@pytest.mark.parametrize(
    ("alertname", "target", "message", "labels", "expected_kind", "expected_mode"),
    [
        (
            "HostBackupFailed",
            "backup",
            "snapshot stale",
            {},
            "backup_health",
            "BackupCheck",
        ),
        (
            "provider_freshness_signal",
            "provider_freshness_signal",
            "last_seen missing",
            {},
            "provider_freshness_signal",
            "ProviderFreshness",
        ),
        (
            "RebootAutoRecoveryActiveBlocker",
            "reboot-auto-recovery-slo",
            "all_required_hosts_not_in_10_minute_reboot_window cpu pressure",
            {"service": "awoooi"},
            "host_recovery",
            "Recover",
        ),
        (
            "PostgresqlDown",
            "postgresql",
            "database unavailable deployment awoooi",
            {"service": "awoooi"},
            "database_health",
            "Status",
        ),
        (
            "AlertChainBroken",
            "alertmanager",
            "monitoring pipeline down awoooi",
            {"service": "awoooi"},
            "alert_chain_health",
            "Status",
        ),
        (
            "HostDown",
            "192.168.0.120",
            "host_down vm_not_running",
            {"host": "192.168.0.120"},
            "host_recovery",
            "Recover",
        ),
        (
            "KubePodCrashLooping",
            "awoooi-api",
            "k3s pod crashloopbackoff",
            {"service": "awoooi"},
            "awoooi_k3s",
            "AwoooRepair",
        ),
    ],
)
def test_agent99_routing_replay_corpus(
    alertname: str,
    target: str,
    message: str,
    labels: dict[str, str],
    expected_kind: str,
    expected_mode: str,
) -> None:
    payload = build_agent99_sre_alert(
        alert_id=f"replay-{alertname}",
        alertname=alertname,
        severity="critical",
        namespace="awoooi-prod",
        target_resource=target,
        message=message,
        labels=labels,
        fingerprint=f"fp-{alertname}",
    )

    assert payload["kind"] == expected_kind
    assert payload["suggestedMode"] == expected_mode


def test_alert_kind_mapping_for_route_and_performance() -> None:
    assert (
        resolve_agent99_alert_kind(
            alertname="PublicRoute502",
            severity="critical",
            namespace="awoooi-prod",
            target_resource="awoooi-web",
            message="HTTP 502",
        )
        == "public_route_502"
    )
    assert (
        resolve_agent99_alert_kind(
            alertname="HostHighCpuLoad",
            severity="warning",
            namespace="node",
            target_resource="192.168.0.110",
            message="load per core is high",
        )
        == "performance_pressure"
    )


def test_write_agent99_sre_alert_to_configured_inbox(tmp_path) -> None:
    payload = build_agent99_sre_alert(
        alert_id="alert-write-smoke",
        alertname="HostHighCpuLoad",
        severity="warning",
        namespace="node",
        target_resource="192.168.0.110",
        message="CPU load per core is high",
        labels={"host": "192.168.0.110"},
    )

    written = write_agent99_sre_alert(payload, inbox_path=str(tmp_path))

    assert written is not None
    data = json.loads(written.read_text(encoding="utf-8"))
    assert data["id"] == "awoooi-alertmanager-alert-write-smoke"
    assert data["kind"] == "performance_pressure"
    assert data["suggestedMode"] == "Perf"


def test_post_agent99_sre_alert_to_relay(monkeypatch) -> None:
    payload = build_agent99_sre_alert(
        alert_id="relay-smoke",
        alertname="provider_freshness_signal",
        severity="warning",
        namespace="awoooi-prod",
        target_resource="provider_freshness_signal",
        message="freshness last_seen missing",
        labels={"service": "awoooi", "authorization": "must-not-leak"},
    )
    seen: dict[str, object] = {}

    class FakeResponse:
        def getcode(self) -> int:
            return 202

        def read(self, _limit: int) -> bytes:
            return b'{"ok":true}'

        def __enter__(self) -> FakeResponse:
            return self

        def __exit__(self, *_args: object) -> None:
            return None

    def fake_urlopen(request, timeout):  # type: ignore[no-untyped-def]
        seen["url"] = request.full_url
        seen["timeout"] = timeout
        seen["headers"] = dict(request.header_items())
        seen["payload"] = json.loads(request.data.decode("utf-8"))
        return FakeResponse()

    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.urllib.request.urlopen",
        fake_urlopen,
    )

    receipt = post_agent99_sre_alert(
        payload,
        relay_url="http://192.168.0.99:8787/agent99/sre-alert/",
        relay_token="relay-token",
        timeout_seconds=1.5,
    )

    assert receipt == {"status": 202, "body": '{"ok":true}'}
    assert seen["url"] == "http://192.168.0.99:8787/agent99/sre-alert/"
    assert seen["timeout"] == 1.5
    assert seen["payload"]["kind"] == "provider_freshness_signal"
    assert seen["payload"]["suggestedMode"] == "ProviderFreshness"
    assert "authorization=[redacted]" in seen["payload"]["labels"]
    assert "must-not-leak" not in json.dumps(seen["payload"])
    assert seen["headers"]["X-agent99-relay-token"] == "relay-token"


def test_agent99_single_flight_key_does_not_expose_fingerprint() -> None:
    key = agent99_sre_single_flight_key("private/source/fingerprint")

    assert key.startswith("agent99:sre_dispatch:single_flight:")
    assert "private" not in key


@pytest.mark.asyncio
async def test_agent99_bridge_suppresses_duplicate_single_flight(monkeypatch) -> None:
    async def duplicate(*_args, **_kwargs) -> bool:
        return False

    dispatched: list[dict[str, object]] = []
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.try_acquire_agent99_sre_single_flight",
        duplicate,
    )
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert",
        lambda payload: dispatched.append(payload) or "relay",
    )

    result = await bridge_alertmanager_to_agent99(
        alert_id="duplicate-alert",
        alertname="HostDown",
        severity="critical",
        namespace="node",
        target_resource="192.168.0.120",
        message="host_down",
        fingerprint="same-fingerprint",
    )

    assert result["status"] == "suppressed"
    assert result["reason"] == "single_flight_duplicate"
    assert dispatched == []


@pytest.mark.asyncio
async def test_agent99_bridge_dispatches_single_flight_winner(monkeypatch) -> None:
    async def winner(*_args, **_kwargs) -> bool:
        return True

    dispatched: list[dict[str, object]] = []
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.try_acquire_agent99_sre_single_flight",
        winner,
    )
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert",
        lambda payload: dispatched.append(payload) or "relay",
    )

    result = await bridge_alertmanager_to_agent99(
        alert_id="winner-alert",
        alertname="HostDown",
        severity="critical",
        namespace="node",
        target_resource="192.168.0.120",
        message="host_down",
        fingerprint="winner-fingerprint",
    )

    assert result["status"] == "dispatched"
    assert result["dispatch"] == "relay"
    assert result["suggestedMode"] == "Recover"
    assert len(dispatched) == 1


@pytest.mark.asyncio
async def test_agent99_bridge_releases_lock_when_transport_fails(monkeypatch) -> None:
    async def winner(*_args, **_kwargs) -> bool:
        return True

    released: list[tuple[str, str]] = []

    async def release(fingerprint: str, alert_id: str) -> None:
        released.append((fingerprint, alert_id))

    def fail_transport(_payload: dict[str, object]) -> str:
        raise RuntimeError("relay unavailable")

    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.try_acquire_agent99_sre_single_flight",
        winner,
    )
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.release_agent99_sre_single_flight",
        release,
    )
    monkeypatch.setattr(
        "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert",
        fail_transport,
    )

    result = await bridge_alertmanager_to_agent99(
        alert_id="transport-failure",
        alertname="HostDown",
        severity="critical",
        namespace="node",
        target_resource="192.168.0.120",
        message="host_down",
        fingerprint="transport-fingerprint",
    )

    assert result == {"status": "failed", "reason": "RuntimeError"}
    assert released == [("transport-fingerprint", "transport-failure")]


@pytest.mark.asyncio
async def test_agent99_single_flight_uses_redis_nx(monkeypatch) -> None:
    seen: dict[str, object] = {}

    class FakeRedis:
        async def set(self, key, value, *, ex, nx):  # type: ignore[no-untyped-def]
            seen.update({"key": key, "value": value, "ex": ex, "nx": nx})
            return True

    monkeypatch.setattr("src.core.redis_client.get_redis", lambda: FakeRedis())

    from src.services.agent99_sre_bridge import try_acquire_agent99_sre_single_flight

    acquired = await try_acquire_agent99_sre_single_flight(
        "source-fingerprint", "alert-1"
    )

    assert acquired is True
    assert str(seen["key"]).startswith("agent99:sre_dispatch:single_flight:")
    assert seen["value"] == "alert-1"
    assert seen["ex"] == 600
    assert seen["nx"] is True


@pytest.mark.asyncio
async def test_agent99_single_flight_release_is_owner_checked(monkeypatch) -> None:
    seen: dict[str, object] = {}

    class FakeRedis:
        async def eval(self, script, key_count, key, owner):  # type: ignore[no-untyped-def]
            seen.update(
                {"script": script, "key_count": key_count, "key": key, "owner": owner}
            )
            return 1

    monkeypatch.setattr("src.core.redis_client.get_redis", lambda: FakeRedis())

    await release_agent99_sre_single_flight("source-fingerprint", "alert-1")

    assert "redis.call('get'" in str(seen["script"])
    assert seen["key_count"] == 1
    assert seen["owner"] == "alert-1"