49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
import yaml
|
|
|
|
|
|
VM_URL = os.environ.get("VM_URL", "http://victoria-metrics-single-server:8428").rstrip("/")
|
|
CONFIG_PATH = Path(__file__).with_name("config.yaml")
|
|
|
|
|
|
def _load_config() -> dict:
|
|
with CONFIG_PATH.open("r", encoding="utf-8") as handle:
|
|
return yaml.safe_load(handle) or {}
|
|
|
|
|
|
def _query(promql: str) -> list[dict]:
|
|
response = requests.get(f"{VM_URL}/api/v1/query", params={"query": promql}, timeout=10)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return payload.get("data", {}).get("result", [])
|
|
|
|
|
|
def test_glue_metrics_present():
|
|
series = _query('kube_cronjob_labels{label_atlas_bstein_dev_glue="true"}')
|
|
assert series, "No glue cronjob label series found"
|
|
|
|
|
|
def test_glue_metrics_success_join():
|
|
query = (
|
|
"kube_cronjob_status_last_successful_time "
|
|
'and on(namespace,cronjob) kube_cronjob_labels{label_atlas_bstein_dev_glue="true"}'
|
|
)
|
|
series = _query(query)
|
|
assert series, "No glue cronjob last success series found"
|
|
|
|
|
|
def test_ariadne_schedule_metrics_present():
|
|
cfg = _load_config()
|
|
expected = cfg.get("ariadne_schedule_tasks", [])
|
|
if not expected:
|
|
return
|
|
series = _query("ariadne_schedule_next_run_timestamp_seconds")
|
|
tasks = {item.get("metric", {}).get("task") for item in series}
|
|
missing = [task for task in expected if task not in tasks]
|
|
assert not missing, f"Missing Ariadne schedule metrics for: {', '.join(missing)}"
|