Codex Release Bot
32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
"""Fleet trend alert window bucketing."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from server.utils.fleet_alert_buckets import attach_alerts_to_fleet_points
|
|
|
|
|
|
class _Alert:
|
|
def __init__(self, created_at, **kwargs):
|
|
self.created_at = created_at
|
|
self.server_name = kwargs.get("server_name", "srv")
|
|
self.server_id = kwargs.get("server_id", 1)
|
|
self.alert_type = kwargs.get("alert_type", "cpu")
|
|
self.value = kwargs.get("value", "90")
|
|
self.is_recovery = kwargs.get("is_recovery", False)
|
|
|
|
|
|
def test_attach_alerts_to_sample_window():
|
|
points = [
|
|
{"recorded_at": "2026-06-09 10:00:00", "cpu_avg": 10},
|
|
{"recorded_at": "2026-06-09 10:10:00", "cpu_avg": 20},
|
|
]
|
|
alerts = [
|
|
_Alert(datetime(2026, 6, 9, 10, 5, tzinfo=timezone.utc), alert_type="cpu"),
|
|
_Alert(datetime(2026, 6, 9, 10, 8, tzinfo=timezone.utc), alert_type="mem"),
|
|
]
|
|
out = attach_alerts_to_fleet_points(points, alerts, interval_minutes=10)
|
|
assert out[0]["alert_count"] == 0
|
|
assert out[1]["alert_count"] == 2
|
|
types = {a["alert_type"] for a in out[1]["alerts"]}
|
|
assert types == {"cpu", "mem"}
|