feat(agent99): bridge sre alerts to windows relay

This commit is contained in:
ogt
2026-07-10 00:26:52 +08:00
parent e150a85a6d
commit 8a766db5e2
19 changed files with 6673 additions and 0 deletions

View File

@@ -0,0 +1,143 @@
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"