b18a6c039e
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m43s
CD Pipeline / build-and-deploy (push) Successful in 8m55s
CD Pipeline / post-deploy-checks (push) Successful in 1m50s
E2E Health Check / e2e-health (push) Successful in 46s
AI 技術雷達監控 / ai-technology-watch (push) Failing after 14m55s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 22s
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
"""Regression coverage for PostgreSQL slow-query alert automation routing."""
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import yaml
|
|
|
|
from src.services.awooop_ansible_audit_service import (
|
|
build_ansible_decision_audit_payload,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
RULE_FILES = (
|
|
ROOT / "ops" / "monitoring" / "alerts.yml",
|
|
ROOT / "ops" / "monitoring" / "alerts-unified.yml",
|
|
)
|
|
|
|
|
|
def _find_rule(path: Path, alertname: str) -> dict:
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
for group in payload.get("groups", []):
|
|
for rule in group.get("rules", []):
|
|
if rule.get("alert") == alertname:
|
|
return rule
|
|
raise AssertionError(f"missing {alertname} in {path}")
|
|
|
|
|
|
def _incident(alert_name: str, labels: dict[str, str]) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
incident_id="INC-20260711-SEMANTIC",
|
|
project_id="awoooi",
|
|
alert_category="database_performance",
|
|
notification_type="TYPE-3",
|
|
severity=SimpleNamespace(value="P2"),
|
|
affected_services=["postgres"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
alert_name=alert_name,
|
|
labels=labels,
|
|
annotations={"message": "PostgreSQL has a transaction over 60s"},
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def test_backup_clients_are_excluded_from_slow_query_alert_mirrors() -> None:
|
|
expressions = []
|
|
for path in RULE_FILES:
|
|
rule = _find_rule(path, "PostgreSQLSlowQueries")
|
|
expressions.append(rule["expr"])
|
|
assert "application_name" in rule["expr"]
|
|
assert "pg_dump" in rule["expr"]
|
|
assert "pg_basebackup" in rule["expr"]
|
|
|
|
assert len(set(expressions)) == 1
|
|
|
|
|
|
def test_slow_query_cannot_activate_momo_backup_or_ai_web_catalog() -> None:
|
|
incident = _incident(
|
|
"PostgreSQLSlowQueries",
|
|
{
|
|
"alertname": "PostgreSQLSlowQueries",
|
|
"application_name": "pg_dump",
|
|
"component": "postgres",
|
|
"host": "188",
|
|
"instance": "192.168.0.188:9187",
|
|
"datname": "awoooi_prod",
|
|
},
|
|
)
|
|
|
|
payload = build_ansible_decision_audit_payload(
|
|
incident=incident,
|
|
proposal_data={
|
|
"source": "truth_chain_candidate_backfill",
|
|
"risk_level": "medium",
|
|
"action": "enqueue_allowlisted_ansible_check_mode",
|
|
},
|
|
decision_path="repair_candidate_controlled_queue",
|
|
not_used_reason="candidate backfill",
|
|
)
|
|
|
|
assert payload is None
|
|
|
|
|
|
def test_momo_backup_failure_still_activates_bounded_backup_catalog() -> None:
|
|
incident = _incident(
|
|
"MomoPostgresBackupFailed",
|
|
{
|
|
"alertname": "MomoPostgresBackupFailed",
|
|
"component": "postgres",
|
|
"host": "188",
|
|
"job": "momo-pg-backup",
|
|
},
|
|
)
|
|
|
|
payload = build_ansible_decision_audit_payload(
|
|
incident=incident,
|
|
proposal_data={
|
|
"source": "truth_chain_candidate_backfill",
|
|
"risk_level": "low",
|
|
"action": "enqueue_allowlisted_ansible_check_mode",
|
|
},
|
|
decision_path="repair_candidate_controlled_queue",
|
|
not_used_reason="candidate backfill",
|
|
)
|
|
|
|
assert payload is not None
|
|
candidate = payload["input"]["executor_candidates"][0]
|
|
assert candidate["catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert candidate["semantic_match"] is True
|
|
assert "momopostgresbackupfailed" in candidate[
|
|
"matched_activation_keywords"
|
|
]
|