prole/tests/installer/test_policy.py

from __future__ import annotations


import json


def _kubectl_nodes_json(*, items: list[dict]) -> str:
    return json.dumps({"apiVersion": "v1", "items": items})


def test_evaluate_optional_workloads_allowed_true_two_ready(monkeypatch):
    from knoe.core import policy

    class _Res:
        returncode = 0
        stderr = ""
        stdout = _kubectl_nodes_json(
            items=[
                {
                    "spec": {},
                    "status": {
                        "conditions": [
                            {"type": "Ready", "status": "True"},
                        ]
                    },
                },
                {
                    "spec": {},
                    "status": {
                        "conditions": [
                            {"type": "Ready", "status": "True"},
                        ]
                    },
                },
            ]
        )

    def _run(*_args, **_kwargs):
        return _Res()

    monkeypatch.setattr(policy.subprocess, "run", _run)

    allowed, ready_schedulable, reason = policy.evaluate_optional_workloads_allowed(
        min_nodes=2
    )
    assert allowed is True
    assert ready_schedulable == 2
    assert "optional_workloads_allowed=True" in reason
    assert "min_required=2" in reason


def test_evaluate_optional_workloads_allowed_false_one_ready(monkeypatch):
    from knoe.core import policy

    class _Res:
        returncode = 0
        stderr = ""
        stdout = _kubectl_nodes_json(
            items=[
                {
                    "spec": {},
                    "status": {
                        "conditions": [
                            {"type": "Ready", "status": "True"},
                        ]
                    },
                }
            ]
        )

    monkeypatch.setattr(policy.subprocess, "run", lambda *_a, **_k: _Res())

    allowed, ready_schedulable, reason = policy.evaluate_optional_workloads_allowed(
        min_nodes=2
    )
    assert allowed is False
    assert ready_schedulable == 1
    assert "optional_workloads_allowed=False" in reason
    assert "min_required=2" in reason


def test_evaluate_optional_workloads_ignores_unschedulable(monkeypatch):
    from knoe.core import policy

    class _Res:
        returncode = 0
        stderr = ""
        stdout = _kubectl_nodes_json(
            items=[
                {
                    "spec": {"unschedulable": True},
                    "status": {
                        "conditions": [
                            {"type": "Ready", "status": "True"},
                        ]
                    },
                },
                {
                    "spec": {},
                    "status": {
                        "conditions": [
                            {"type": "Ready", "status": "True"},
                        ]
                    },
                },
            ]
        )

    monkeypatch.setattr(policy.subprocess, "run", lambda *_a, **_k: _Res())

    allowed, ready_schedulable, _reason = policy.evaluate_optional_workloads_allowed(
        min_nodes=2
    )
    assert allowed is False
    assert ready_schedulable == 1


def test_evaluate_optional_workloads_kubectl_error_is_disallowed(monkeypatch):
    from knoe.core import policy

    class _Res:
        returncode = 1
        stderr = "boom"
        stdout = ""

    monkeypatch.setattr(policy.subprocess, "run", lambda *_a, **_k: _Res())

    allowed, ready_schedulable, reason = policy.evaluate_optional_workloads_allowed(
        min_nodes=2
    )
    assert allowed is False
    assert ready_schedulable == 0
    assert "failed to query nodes" in reason