fix(monitoring): clarify alert chain smoke evidence

apps/api/tests/test_alert_chain_smoke_metric.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import time
+import unittest
+from pathlib import Path
+
+
+SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "alert_chain_smoke_test.py"
+SPEC = importlib.util.spec_from_file_location("alert_chain_smoke_test", SCRIPT_PATH)
+alert_chain_smoke_test = importlib.util.module_from_spec(SPEC)
+assert SPEC and SPEC.loader
+sys.dont_write_bytecode = True
+sys.modules[SPEC.name] = alert_chain_smoke_test
+SPEC.loader.exec_module(alert_chain_smoke_test)
+
+
+class AlertChainSmokeMetricTest(unittest.TestCase):
+    def test_parse_app_alert_chain_metric_samples(self):
+        samples = alert_chain_smoke_test.parse_app_alert_chain_metric_samples(
+            "\n".join([
+                "# HELP awoooi_alert_chain_last_success_timestamp Last successful alert chain",
+                'awoooi_alert_chain_last_success_timestamp{source="alertmanager"} 123.5',
+                'awoooi_alert_chain_last_success_timestamp{source="sentry"} 120',
+                "unrelated_metric 1",
+            ])
+        )
+
+        self.assertEqual(
+            samples,
+            [
+                alert_chain_smoke_test.AlertChainMetricSample(
+                    source="alertmanager",
+                    timestamp=123.5,
+                    evidence_path="app_metrics",
+                ),
+                alert_chain_smoke_test.AlertChainMetricSample(
+                    source="sentry",
+                    timestamp=120.0,
+                    evidence_path="app_metrics",
+                ),
+            ],
+        )
+
+    def test_newest_sample_for_source_prefers_requested_source(self):
+        samples = [
+            alert_chain_smoke_test.AlertChainMetricSample("sentry", 999.0, "prometheus"),
+            alert_chain_smoke_test.AlertChainMetricSample("alertmanager", 100.0, "prometheus"),
+            alert_chain_smoke_test.AlertChainMetricSample("alertmanager", 200.0, "app_metrics"),
+        ]
+
+        sample = alert_chain_smoke_test._newest_sample_for_source(samples, "alertmanager")
+
+        self.assertEqual(sample.timestamp, 200.0)
+        self.assertEqual(sample.evidence_path, "app_metrics")
+
+    def test_alert_chain_metric_result_marks_recent_app_metric_as_scrape_delay(self):
+        sample = alert_chain_smoke_test.AlertChainMetricSample(
+            source="alertmanager",
+            timestamp=time.time() - 60,
+            evidence_path="app_metrics",
+        )
+
+        result = alert_chain_smoke_test._alert_chain_metric_result(sample, fallback=True)
+
+        self.assertTrue(result.passed)
+        self.assertIn("Prometheus scrape 尚未看到", result.message)
+
+    def test_alert_chain_metric_result_fails_persistent_silence(self):
+        sample = alert_chain_smoke_test.AlertChainMetricSample(
+            source="alertmanager",
+            timestamp=time.time() - alert_chain_smoke_test.MAX_ALERT_CHAIN_SILENCE_SECONDS - 60,
+            evidence_path="prometheus",
+        )
+
+        result = alert_chain_smoke_test._alert_chain_metric_result(sample)
+
+        self.assertFalse(result.passed)
+        self.assertTrue(result.critical)
+
+
+if __name__ == "__main__":
+    unittest.main()