598f33ae8b
All checks were successful
Code Review / ai-code-review (push) Successful in 11s
Deploy Alert Rules / Deploy Prometheus Alert Rules (push) Successful in 22s
CD Pipeline / tests (push) Successful in 3m55s
CD Pipeline / build-and-deploy (push) Successful in 3m31s
CD Pipeline / post-deploy-checks (push) Successful in 1m33s
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
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()
|