ariadne/tests/test_k8s_pods.py

from __future__ import annotations

import pytest

from ariadne.k8s import pods as pods_module


def test_list_pods_encodes_selector(monkeypatch) -> None:
    captured = {}

    def fake_get_json(path: str):
        captured["path"] = path
        return {"items": []}

    monkeypatch.setattr(pods_module, "get_json", fake_get_json)
    pods_module.list_pods("demo", "app=nextcloud")
    assert "labelSelector=app%3Dnextcloud" in captured["path"]


def test_list_pods_rejects_missing_namespace() -> None:
    with pytest.raises(pods_module.PodSelectionError, match="namespace missing"):
        pods_module.list_pods(" ", "app=nextcloud")


def test_parse_start_time_handles_empty_invalid_and_naive_values() -> None:
    assert pods_module._parse_start_time(None) == 0.0
    assert pods_module._parse_start_time("not-a-date") == 0.0
    assert pods_module._parse_start_time("2026-01-20T00:00:00") > 0


def test_ready_helper_handles_malformed_conditions() -> None:
    assert pods_module._is_ready({"status": {"phase": "Running"}}) is False
    assert pods_module._is_ready({"status": {"phase": "Running", "conditions": [None]}}) is False
    assert pods_module._is_ready({"status": {"phase": "Running", "conditions": [{"type": "ContainersReady"}]}}) is False


def test_select_pod_picks_ready_latest(monkeypatch) -> None:
    payload = {
        "items": [
            {
                "metadata": {"name": "old-pod"},
                "status": {
                    "phase": "Running",
                    "startTime": "2026-01-19T00:00:00Z",
                    "conditions": [{"type": "Ready", "status": "True"}],
                },
            },
            {
                "metadata": {"name": "new-pod"},
                "status": {
                    "phase": "Running",
                    "startTime": "2026-01-20T00:00:00Z",
                    "conditions": [{"type": "Ready", "status": "True"}],
                },
            },
        ]
    }

    monkeypatch.setattr(pods_module, "get_json", lambda *_args, **_kwargs: payload)
    pod = pods_module.select_pod("demo", "app=test")
    assert pod.name == "new-pod"


def test_select_pod_ignores_non_ready(monkeypatch) -> None:
    payload = {
        "items": [
            {
                "metadata": {"name": "pending"},
                "status": {"phase": "Pending"},
            },
        ]
    }
    monkeypatch.setattr(pods_module, "get_json", lambda *_args, **_kwargs: payload)

    with pytest.raises(pods_module.PodSelectionError):
        pods_module.select_pod("demo", "app=test")


def test_select_pod_skips_deleting_and_blank_names(monkeypatch) -> None:
    payload = {
        "items": [
            {
                "metadata": {"name": "deleting", "deletionTimestamp": "2026-01-20T00:00:00Z"},
                "status": {"phase": "Running", "conditions": [{"type": "Ready", "status": "True"}]},
            },
            {
                "metadata": {"name": " "},
                "status": {"phase": "Running", "conditions": [{"type": "Ready", "status": "True"}]},
            },
            {
                "metadata": {"name": "ready"},
                "status": {"phase": "Running", "nodeName": "titan-1", "conditions": [{"type": "Ready", "status": "True"}]},
            },
        ]
    }
    monkeypatch.setattr(pods_module, "get_json", lambda *_args, **_kwargs: payload)

    pod = pods_module.select_pod("demo", "app=test")

    assert pod.name == "ready"
    assert pod.node == "titan-1"