import re
from pathlib import Path

import yaml


REPO_ROOT = Path(__file__).resolve().parents[1]
ALERTING_DIR = REPO_ROOT / "configs" / "grafana" / "provisioning" / "alerting"

EXPECTED_THRESHOLDS = {
    "airflow-alert-rules.yml": {
        "airflow_scheduler_down": ("lt", [0.1]),
        "airflow_queue_high": ("gt", [50]),
        "airflow_task_failures_high": ("gt", [0.1]),
        "airflow_parse_time_high": ("gt", [30]),
    },
    "kafka-alert-rules.yml": {
        "kafka_broker_down": ("lt", [1]),
        "kafka_consumer_lag_high": ("gt", [10000]),
        "kafka_no_messages_produced": ("lt", [0.1]),
    },
}


def _rules_by_uid(path):
    document = yaml.safe_load(path.read_text(encoding="utf-8"))
    return {
        rule["uid"]: rule
        for group in document["groups"]
        for rule in group["rules"]
    }


def test_range_alerts_reduce_series_before_threshold():
    """Диапазонный запрос сводится к числу до сравнения с порогом."""
    for filename, expected_thresholds in EXPECTED_THRESHOLDS.items():
        rules = _rules_by_uid(ALERTING_DIR / filename)

        assert set(rules) == set(expected_thresholds)

        for uid, (comparison, params) in expected_thresholds.items():
            rule = rules[uid]
            stages = {stage["refId"]: stage for stage in rule["data"]}

            threshold = stages[rule["condition"]]["model"]
            assert threshold["type"] == "threshold", uid

            reduce_stage = stages[threshold["expression"]]["model"]
            assert reduce_stage["type"] == "reduce", uid
            assert reduce_stage["reducer"] == "last", uid

            query = stages[reduce_stage["expression"]]["model"]
            assert query["range"] is True, uid

            evaluator = threshold["conditions"][0]["evaluator"]
            assert evaluator == {"type": comparison, "params": params}, uid


def test_alert_annotations_use_reduced_values():
    """Описание алерта выводит число после Reduce, а не диапазонный ряд."""
    for filename in EXPECTED_THRESHOLDS:
        for uid, rule in _rules_by_uid(ALERTING_DIR / filename).items():
            stages = {stage["refId"]: stage for stage in rule["data"]}

            for annotation in rule.get("annotations", {}).values():
                for ref_id in re.findall(r"\$values\.([A-Z])", annotation):
                    model = stages[ref_id]["model"]
                    assert model.get("type") == "reduce", uid