81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from ariadne.services import platform_quality_probe as module
|
|
|
|
|
|
def _settings() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
platform_quality_probe_namespace="monitoring",
|
|
platform_quality_probe_script_configmap="platform-quality-suite-probe-script",
|
|
platform_quality_probe_image="curlimages/curl:8.12.1",
|
|
platform_quality_probe_job_ttl_sec=1800,
|
|
platform_quality_probe_wait_timeout_sec=20.0,
|
|
platform_quality_probe_pushgateway_url="http://platform-quality-gateway.monitoring.svc.cluster.local:9091",
|
|
platform_quality_probe_http_timeout_sec=12,
|
|
)
|
|
|
|
|
|
def test_payload_matches_expected_contract(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
payload = module.PlatformQualityProbeService()._job_payload("probe-job")
|
|
|
|
assert payload["metadata"]["namespace"] == "monitoring"
|
|
assert payload["metadata"]["labels"]["atlas.bstein.dev/trigger"] == "ariadne"
|
|
container = payload["spec"]["template"]["spec"]["containers"][0]
|
|
assert container["image"] == "curlimages/curl:8.12.1"
|
|
assert container["command"] == ["/bin/sh", "/scripts/platform_quality_suite_probe.sh"]
|
|
assert payload["spec"]["template"]["spec"]["volumes"][0]["configMap"]["name"] == "platform-quality-suite-probe-script"
|
|
|
|
|
|
def test_run_wait_success(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1720000000)
|
|
|
|
posted: dict[str, object] = {}
|
|
|
|
def fake_post(path: str, payload: dict[str, object]) -> dict[str, object]:
|
|
posted["path"] = path
|
|
posted["payload"] = payload
|
|
return {"metadata": {"name": "probe-job-1"}}
|
|
|
|
calls = iter(
|
|
[
|
|
{"status": {"active": 1}},
|
|
{"status": {"succeeded": 1}},
|
|
]
|
|
)
|
|
|
|
monkeypatch.setattr(module, "post_json", fake_post)
|
|
monkeypatch.setattr(module, "get_json", lambda _path: next(calls))
|
|
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
|
|
|
result = module.PlatformQualityProbeService().run(wait=True)
|
|
|
|
assert posted["path"] == "/apis/batch/v1/namespaces/monitoring/jobs"
|
|
assert result == {"job": "probe-job-1", "status": "ok"}
|
|
|
|
|
|
def test_run_wait_failure_raises(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1720000000)
|
|
monkeypatch.setattr(module, "post_json", lambda _path, _payload: {"metadata": {"name": "probe-job-2"}})
|
|
monkeypatch.setattr(module, "get_json", lambda _path: {"status": {"failed": 1}})
|
|
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
|
|
|
with pytest.raises(RuntimeError, match="platform quality probe job probe-job-2 error"):
|
|
module.PlatformQualityProbeService().run(wait=True)
|
|
|
|
|
|
def test_run_without_wait_queues(monkeypatch) -> None:
|
|
monkeypatch.setattr(module, "settings", _settings())
|
|
monkeypatch.setattr(module.time, "time", lambda: 1720000000)
|
|
monkeypatch.setattr(module, "post_json", lambda _path, _payload: {"metadata": {"name": "probe-job-3"}})
|
|
|
|
result = module.PlatformQualityProbeService().run(wait=False)
|
|
|
|
assert result == {"job": "probe-job-3", "status": "queued"}
|