30 lines
912 B
Python
30 lines
912 B
Python
from __future__ import annotations
|
|
|
|
from ariadne.services import cluster_state_fetchers as fetchers
|
|
|
|
|
|
def test_cluster_state_fetchers_record_individual_api_errors(monkeypatch) -> None:
|
|
def fail_json(path: str):
|
|
raise RuntimeError(f"boom {path}")
|
|
|
|
monkeypatch.setattr(fetchers, "_get_json", fail_json)
|
|
errors: list[str] = []
|
|
|
|
assert fetchers._fetch_nodes(errors) == ({}, [], {})
|
|
assert fetchers._fetch_flux(errors) == {}
|
|
assert fetchers._fetch_pods(errors) == ([], [], [], [], {})
|
|
assert fetchers._fetch_jobs(errors) == {}
|
|
assert fetchers._fetch_longhorn(errors) == {}
|
|
assert fetchers._fetch_workload_health(errors) == {}
|
|
assert fetchers._fetch_events(errors) == {}
|
|
assert [entry.split(":", 1)[0] for entry in errors] == [
|
|
"nodes",
|
|
"flux",
|
|
"pods",
|
|
"jobs",
|
|
"longhorn",
|
|
"workloads_health",
|
|
"events",
|
|
]
|
|
|