144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from src.services.agent99_sre_bridge import (
|
|
build_agent99_sre_alert,
|
|
post_agent99_sre_alert,
|
|
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 True
|
|
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"]
|
|
|
|
|
|
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"
|