ariadne/tests/test_image_sweeper.py

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"


def test_image_sweeper_run_error(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": {"failed": 1}}

    monkeypatch.setattr(sweeper_module, "post_json", fake_post)
    monkeypatch.setattr(sweeper_module, "get_json", fake_get)

    svc = ImageSweeperService()
    try:
        svc.run(wait=True)
    except RuntimeError as exc:
        assert "image sweeper job" in str(exc)
    else:
        raise AssertionError("expected image sweeper failure to raise")