fix(backup): preserve alert receipt identity

scripts/ops/tests/test_backup_alert_live_visibility.py

@@ -0,0 +1,90 @@
+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)

scripts/ops/tests/test_backup_health_textfile_exporter.py

@@ -248,6 +248,45 @@ def test_backup_alert_receipt_metrics_require_all_stages(tmp_path: Path, monkeyp
     )
 
 
+def test_backup_schedule_singleton_receipt_is_required_without_false_receipt(
+    tmp_path: Path, monkeypatch
+) -> None:
+    exporter = load_exporter()
+    receipt_dir = tmp_path / "alert-receipts"
+    receipt_dir.mkdir()
+    now = 1_782_900_000
+
+    monkeypatch.setattr(exporter, "BACKUP_ALERT_RECEIPT_DIR", receipt_dir)
+    monkeypatch.setattr(exporter, "BACKUP_ALERT_RECEIPT_MAX_AGE_HOURS", 168)
+    monkeypatch.setattr(exporter.time, "time", lambda: now)
+
+    requirement = next(
+        row
+        for row in exporter.BACKUP_ALERT_RECEIPT_REQUIREMENTS
+        if row["alertname"] == "BackupScheduleSingletonMismatch"
+    )
+    assert requirement == {
+        "host": "110",
+        "alertname": "BackupScheduleSingletonMismatch",
+        "scope": "schedule",
+        "notification_type": "TYPE-1",
+    }
+
+    rendered = "\n".join(exporter._backup_alert_receipt_metric_lines("110"))
+    label_prefix = (
+        'host="110",required_alert="BackupScheduleSingletonMismatch",scope="schedule",'
+        'required_notification_type="TYPE-1",receipt_channel="telegram_awooop",max_age_hours="168"'
+    )
+    assert f"awoooi_backup_alert_receipt_expected_info{{{label_prefix}}} 1" in rendered
+    assert f"awoooi_backup_alert_receipt_fresh{{{label_prefix}}} 0" in rendered
+    for stage in exporter.BACKUP_ALERT_RECEIPT_STAGES:
+        assert (
+            f"awoooi_backup_alert_receipt_stage_fresh{{{label_prefix},stage=\"{stage}\"}} 0"
+            in rendered
+        )
+    assert list(receipt_dir.iterdir()) == []
+
+
 def test_backup_alert_receipt_shared_marker_can_satisfy_all_stages(tmp_path: Path, monkeypatch) -> None:
     exporter = load_exporter()
     receipt_dir = tmp_path / "alert-receipts"