Files
awoooi/scripts/ops/tests/test_backup_alert_live_visibility.py
ogt a535fa652f
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m22s
CD Pipeline / build-and-deploy (push) Successful in 11m52s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 0s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 21s
CD Pipeline / post-deploy-checks (push) Successful in 2m29s
fix(backup): preserve alert receipt identity
2026-07-15 01:08:44 +08:00

91 lines
2.7 KiB
Python

from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
SCRIPT_ROOT = Path(__file__).resolve().parents[1]
CHECK_PATH = SCRIPT_ROOT / "backup-alert-live-visibility-check.py"
def load_check():
spec = importlib.util.spec_from_file_location(
"backup_alert_live_visibility_check",
CHECK_PATH,
)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def test_receipt_verifier_preserves_each_required_alert_identity(monkeypatch) -> None:
check = load_check()
observed: dict[str, object] = {}
def fake_query(base_url: str, expr: str, timeout: int):
observed.update(base_url=base_url, expr=expr, timeout=timeout)
return [
{
"metric": {
"alertname": "BackupAlertReceiptStageMissing",
"alertstate": "firing",
"host": "110",
"required_alert": "BackupScheduleSingletonMismatch",
},
"value": [1_783_000_000, "1"],
},
{
"metric": {
"alertname": "BackupAlertReceiptStageMissing",
"alertstate": "firing",
"host": "110",
"required_alert": "BackupRetentionPolicyNotLatestOnly",
},
"value": [1_783_000_000, "1"],
},
]
monkeypatch.setattr(check, "_prom_query", fake_query)
required = check._receipt_stage_required_alerts("http://prometheus", "110", 8)
assert observed == {
"base_url": "http://prometheus",
"expr": (
'ALERTS{alertname="BackupAlertReceiptStageMissing",alertstate="firing",'
'host="110"}'
),
"timeout": 8,
}
assert [alert.labels["required_alert"] for alert in required] == [
"BackupRetentionPolicyNotLatestOnly",
"BackupScheduleSingletonMismatch",
]
assert all("receipt_channel" not in alert.labels for alert in required)
def test_receipt_verifier_rejects_meta_alert_without_required_alert(monkeypatch) -> None:
check = load_check()
monkeypatch.setattr(
check,
"_prom_query",
lambda *_args: [
{
"metric": {
"alertname": "BackupAlertReceiptStageMissing",
"alertstate": "firing",
"host": "110",
},
"value": [1_783_000_000, "1"],
}
],
)
with pytest.raises(check.VisibilityError, match="required_alert"):
check._receipt_stage_required_alerts("http://prometheus", "110", 8)