test: verify backup monitoring implementation

This commit is contained in:
2026-05-11 21:57:31 +02:00
parent 322b3d1a3b
commit a748512d22
6 changed files with 21 additions and 8 deletions
@@ -1,5 +1,6 @@
from datetime import datetime
from typing import Any
from pydantic import BaseModel, Field, field_validator
@@ -24,7 +24,10 @@ def generate_alerts_for_run(
# 2. Anomaly size alert
bytes_transferred = run.get("bytes_transferred")
if bytes_transferred is not None and previous_runs:
successful_runs = [r for r in previous_runs if r["status"] == "success" and r.get("bytes_transferred") is not None]
successful_runs = [
r for r in previous_runs
if r["status"] == "success" and r.get("bytes_transferred") is not None
]
if len(successful_runs) >= 3:
sizes = [r["bytes_transferred"] for r in successful_runs[-7:]]
median_size = statistics.median(sizes)
@@ -44,7 +47,10 @@ def generate_alerts_for_run(
"run_id": run["id"],
"alert_type": "anomaly_size",
"severity": "warning",
"message": f"Backup job '{job_name}' size anomaly: {bytes_transferred} bytes (median: {median_size})",
"message": (
f"Backup job '{job_name}' size anomaly: "
f"{bytes_transferred} bytes (median: {median_size})"
),
})
# 3. Anomaly duration alert
@@ -60,7 +66,10 @@ def generate_alerts_for_run(
"run_id": run["id"],
"alert_type": "anomaly_duration",
"severity": "warning",
"message": f"Backup job '{job_name}' duration anomaly: {duration_ms}ms (median: {median_duration}ms)",
"message": (
f"Backup job '{job_name}' duration anomaly: "
f"{duration_ms}ms (median: {median_duration}ms)"
),
})
return alerts
@@ -105,7 +114,10 @@ def check_missed_schedules(
"run_id": None,
"alert_type": "missed_schedule",
"severity": "warning",
"message": f"Backup job '{job['name']}' missed schedule: last run at {last_run_time} (expected every {interval}s)",
"message": (
f"Backup job '{job['name']}' missed schedule: "
f"last run at {last_run_time} (expected every {interval}s)"
),
})
return alerts