49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import types
|
||
|
|
|
||
|
|
from ariadne.services import image_sweeper as sweeper_module
|
||
|
|
from ariadne.services.image_sweeper import ImageSweeperService
|
||
|
|
|
||
|
|
|
||
|
|
def test_image_sweeper_job_payload(monkeypatch) -> None:
|
||
|
|
dummy_settings = types.SimpleNamespace(
|
||
|
|
image_sweeper_namespace="maintenance",
|
||
|
|
image_sweeper_service_account="node-image-sweeper",
|
||
|
|
image_sweeper_job_ttl_sec=3600,
|
||
|
|
image_sweeper_wait_timeout_sec=30.0,
|
||
|
|
)
|
||
|
|
monkeypatch.setattr(sweeper_module, "settings", dummy_settings)
|
||
|
|
|
||
|
|
svc = ImageSweeperService()
|
||
|
|
payload = svc._job_payload("job-1")
|
||
|
|
|
||
|
|
assert payload["metadata"]["name"] == "job-1"
|
||
|
|
spec = payload["spec"]["template"]["spec"]
|
||
|
|
assert spec["serviceAccountName"] == "node-image-sweeper"
|
||
|
|
assert spec["containers"][0]["env"][0]["value"] == "true"
|
||
|
|
|
||
|
|
|
||
|
|
def test_image_sweeper_run_wait(monkeypatch) -> None:
|
||
|
|
dummy_settings = types.SimpleNamespace(
|
||
|
|
image_sweeper_namespace="maintenance",
|
||
|
|
image_sweeper_service_account="node-image-sweeper",
|
||
|
|
image_sweeper_job_ttl_sec=3600,
|
||
|
|
image_sweeper_wait_timeout_sec=30.0,
|
||
|
|
)
|
||
|
|
monkeypatch.setattr(sweeper_module, "settings", dummy_settings)
|
||
|
|
|
||
|
|
def fake_post(path, payload):
|
||
|
|
assert path.endswith("/maintenance/jobs")
|
||
|
|
return {"metadata": {"name": "job-1"}}
|
||
|
|
|
||
|
|
def fake_get(_path):
|
||
|
|
return {"status": {"succeeded": 1}}
|
||
|
|
|
||
|
|
monkeypatch.setattr(sweeper_module, "post_json", fake_post)
|
||
|
|
monkeypatch.setattr(sweeper_module, "get_json", fake_get)
|
||
|
|
|
||
|
|
svc = ImageSweeperService()
|
||
|
|
result = svc.run(wait=True)
|
||
|
|
assert result["status"] == "ok"
|