94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
|
"""Парсинг Ruler / слияние папок и HTTP-mock синхронизации Grafana."""
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
import respx
|
||
|
|
|
||
|
|
from onguard24.integrations.grafana_topology import (
|
||
|
|
merge_folder_rows,
|
||
|
|
parse_ruler_rules,
|
||
|
|
)
|
||
|
|
from onguard24.modules.grafana_catalog import pull_topology
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_ruler_grafana_managed() -> None:
|
||
|
|
data = {
|
||
|
|
"nginx": [
|
||
|
|
{
|
||
|
|
"name": "prometheus (Nginx)",
|
||
|
|
"interval": "60s",
|
||
|
|
"rules": [
|
||
|
|
{
|
||
|
|
"grafana_alert": {"uid": "uid1", "title": "Nginx Down"},
|
||
|
|
"labels": {"service": "nginx", "severity": "critical"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
rows = parse_ruler_rules(data)
|
||
|
|
assert len(rows) == 1
|
||
|
|
assert rows[0].namespace_uid == "nginx"
|
||
|
|
assert rows[0].rule_uid == "uid1"
|
||
|
|
assert rows[0].rule_group_name == "prometheus (Nginx)"
|
||
|
|
assert rows[0].labels["service"] == "nginx"
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_ruler_skips_non_grafana_alert() -> None:
|
||
|
|
data = {"x": [{"name": "g", "rules": [{"expr": "1"}]}]}
|
||
|
|
assert parse_ruler_rules(data) == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_merge_folder_rows_adds_namespaces() -> None:
|
||
|
|
api = [{"uid": "system", "title": "System", "parentUid": None}]
|
||
|
|
merged = merge_folder_rows(api, {"nginx", "system"})
|
||
|
|
uids = {m[0] for m in merged}
|
||
|
|
assert uids == {"system", "nginx"}
|
||
|
|
titles = {m[0]: m[1] for m in merged}
|
||
|
|
assert titles["nginx"] == "nginx"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
@respx.mock
|
||
|
|
async def test_pull_topology_end_to_end() -> None:
|
||
|
|
base = "https://grafana.example.com"
|
||
|
|
respx.get(f"{base}/api/org").mock(
|
||
|
|
return_value=httpx.Response(200, json={"id": 3, "name": "adibrov"})
|
||
|
|
)
|
||
|
|
def _folders(request: httpx.Request) -> httpx.Response:
|
||
|
|
if "parentUid" in str(request.url):
|
||
|
|
return httpx.Response(200, json=[])
|
||
|
|
return httpx.Response(
|
||
|
|
200,
|
||
|
|
json=[{"uid": "nginx", "title": "Nginx Alerts", "parentUid": None}],
|
||
|
|
)
|
||
|
|
|
||
|
|
respx.get(f"{base}/api/folders").mock(side_effect=_folders)
|
||
|
|
ruler_body = {
|
||
|
|
"nginx": [
|
||
|
|
{
|
||
|
|
"name": "grp",
|
||
|
|
"interval": "1m",
|
||
|
|
"rules": [
|
||
|
|
{
|
||
|
|
"grafana_alert": {"uid": "r1", "title": "Down"},
|
||
|
|
"labels": {"service": "nginx"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
respx.get(f"{base}/api/ruler/grafana/api/v1/rules").mock(
|
||
|
|
return_value=httpx.Response(200, json=ruler_body)
|
||
|
|
)
|
||
|
|
|
||
|
|
out, err = await pull_topology(base, "test-token")
|
||
|
|
assert err is None
|
||
|
|
assert out is not None
|
||
|
|
assert out.org_id == 3
|
||
|
|
assert out.org_name == "adibrov"
|
||
|
|
assert len(out.folder_rows) == 1
|
||
|
|
assert out.folder_rows[0][0] == "nginx"
|
||
|
|
assert len(out.rules) == 1
|
||
|
|
assert out.rules[0].rule_uid == "r1"
|