prole/tests/installer/test_topology.py

from __future__ import annotations

import json
import subprocess

from knoe.core import topology as topology_mod


def _completed(args: list[str], rc: int = 0, out: str = "", err: str = "") -> subprocess.CompletedProcess[str]:
    return subprocess.CompletedProcess(args=args, returncode=rc, stdout=out, stderr=err)


def test_extract_inventory_from_marked_log_payload() -> None:
    payload = {
        "nodeName": "myrddin",
        "cpuModel": "Intel",
        "filesystems": [{"mountpoint": "/synology/d001", "availableBytes": 123}],
    }
    text = (
        "hello\n"
        f"{topology_mod.INVENTORY_BEGIN_MARKER}\n"
        f"{json.dumps(payload)}\n"
        f"{topology_mod.INVENTORY_END_MARKER}\n"
    )
    parsed = topology_mod._extract_inventory_from_log(text)
    assert parsed == payload


def test_merge_node_monitoring_eligible_with_synology_d004() -> None:
    k8s_node = {
        "name": "merlin",
        "roles": ("worker",),
        "labels": {
            "kubernetes.io/arch": "amd64",
            "kubernetes.io/os": "linux",
        },
        "taints": (),
        "ready": True,
        "internal_ip": "10.0.0.11",
        "architecture": "amd64",
        "operating_system": "linux",
        "allocatable_cpu": "4",
        "allocatable_memory_bytes": 16 * 1024 * 1024 * 1024,
        "allocatable_ephemeral_storage_bytes": 100 * 1024 * 1024 * 1024,
    }
    inventory = {
        "filesystems": [
            {
                "device": "/dev/sdb",
                "mountpoint": "/synology/d004",
                "fstype": "ext4",
                "sizeBytes": 512 * 1024 * 1024 * 1024,
                "usedBytes": 128 * 1024 * 1024 * 1024,
                "availableBytes": 384 * 1024 * 1024 * 1024,
                "transport": "sata",
                "rotational": True,
                "storageClass": "synology",
            }
        ]
    }

    merged = topology_mod._merge_node(
        k8s_node,
        inventory=inventory,
        required_monitoring_mountpoints=("/synology/d001",),
        monitoring_min_free_bytes=200 * 1024 * 1024 * 1024,
    )

    assert merged.capabilities.matching_mountpoints == ("/synology/d004",)
    assert merged.capabilities.monitoring_eligible is True


def test_discover_cluster_topology_partial_results_and_xml(tmp_path, monkeypatch) -> None:
    nodes_payload = {
        "items": [
            {
                "metadata": {
                    "name": "myrddin",
                    "labels": {
                        "kubernetes.io/arch": "amd64",
                        "kubernetes.io/os": "linux",
                        "node-role.kubernetes.io/control-plane": "",
                    },
                },
                "spec": {},
                "status": {
                    "conditions": [{"type": "Ready", "status": "True"}],
                    "addresses": [{"type": "InternalIP", "address": "10.0.0.10"}],
                    "allocatable": {
                        "cpu": "4",
                        "memory": "16Gi",
                        "ephemeral-storage": "100Gi",
                    },
                },
            },
            {
                "metadata": {
                    "name": "merlin",
                    "labels": {
                        "kubernetes.io/arch": "amd64",
                        "kubernetes.io/os": "linux",
                    },
                },
                "spec": {},
                "status": {
                    "conditions": [{"type": "Ready", "status": "True"}],
                    "addresses": [{"type": "InternalIP", "address": "10.0.0.11"}],
                    "allocatable": {
                        "cpu": "4",
                        "memory": "16Gi",
                        "ephemeral-storage": "100Gi",
                    },
                },
            },
        ]
    }
    pods_payload = {
        "items": [
            {"metadata": {"name": "collector-myrddin"}, "spec": {"nodeName": "myrddin"}},
            {"metadata": {"name": "collector-merlin"}, "spec": {"nodeName": "merlin"}},
        ]
    }

    def fake_kubectl_json(_base_cmd, args, **_kwargs):
        if args[:2] == ["get", "nodes"]:
            return nodes_payload
        if "pods" in args:
            return pods_payload
        raise AssertionError(f"unexpected kubectl json args: {args}")

    def fake_run_kubectl(_base_cmd, args, **_kwargs):
        if args[:4] == ["-n", "kube-system", "apply", "-f"]:
            return _completed(args)
        if args[:4] == ["-n", "kube-system", "logs", "collector-myrddin"]:
            body = {
                "nodeName": "myrddin",
                "cpuModel": "Intel i7",
                "physicalCores": 6,
                "logicalCores": 12,
                "memoryBytes": 17179869184,
                "filesystems": [
                    {
                        "device": "/dev/nvme0n1p1",
                        "mountpoint": "/synology/d001",
                        "fstype": "ext4",
                        "sizeBytes": 1000,
                        "usedBytes": 100,
                        "availableBytes": 900,
                        "transport": "nvme",
                        "rotational": False,
                        "storageClass": "nvme",
                    }
                ],
            }
            body_json = json.dumps(body, separators=(",", ":"), sort_keys=True)
            text = (
                f"{topology_mod.INVENTORY_BEGIN_MARKER}\n"
                f"{body_json}\n"
                f"{topology_mod.INVENTORY_END_MARKER}\n"
            )
            return _completed(args, out=text)
        if args[:4] == ["-n", "kube-system", "logs", "collector-merlin"]:
            return _completed(args, out="collector still warming up\n")
        if args[:4] == ["-n", "kube-system", "delete", "daemonset"]:
            return _completed(args)
        raise AssertionError(f"unexpected kubectl run args: {args}")

    clock = {"now": 0.0}

    def fake_monotonic():
        clock["now"] += 1.0
        return clock["now"]

    monkeypatch.setattr(topology_mod, "_kubectl_json", fake_kubectl_json)
    monkeypatch.setattr(topology_mod, "_run_kubectl", fake_run_kubectl)
    monkeypatch.setattr(topology_mod.time, "monotonic", fake_monotonic)
    monkeypatch.setattr(topology_mod.time, "sleep", lambda _s: None)

    result = topology_mod.discover_cluster_topology(
        kubectl_base_cmd=["kubectl"],
        topology_root=tmp_path / "topology",
        config=topology_mod.TopologyDiscoveryConfig(
            timeout_seconds=1,
            poll_interval_seconds=0.01,
            monitoring_min_free_bytes=100,
        ),
    )

    assert result.topology.collection_status == "partial"
    assert result.reported_nodes == ("myrddin",)
    assert result.missing_nodes == ("merlin",)

    myrddin = next(node for node in result.topology.nodes if node.name == "myrddin")
    merlin = next(node for node in result.topology.nodes if node.name == "merlin")
    assert myrddin.capabilities.monitoring_eligible is True
    assert merlin.capabilities.monitoring_eligible is None
    assert "topology_unavailable" in merlin.capabilities.reasons

    assert (tmp_path / "topology" / "cluster.xml").is_file()
    assert (tmp_path / "topology" / "nodes" / "myrddin.xml").is_file()
    assert (tmp_path / "topology" / "nodes" / "merlin.xml").is_file()


def test_discover_cluster_topology_failed_collector_keeps_k8s_rows(tmp_path, monkeypatch) -> None:
    nodes_payload = {
        "items": [
            {
                "metadata": {
                    "name": "pi",
                    "labels": {"kubernetes.io/arch": "arm64", "kubernetes.io/os": "linux"},
                },
                "spec": {},
                "status": {
                    "conditions": [{"type": "Ready", "status": "True"}],
                    "addresses": [{"type": "InternalIP", "address": "10.0.0.12"}],
                    "allocatable": {"cpu": "2", "memory": "4Gi", "ephemeral-storage": "20Gi"},
                },
            }
        ]
    }

    def fake_kubectl_json(_base_cmd, args, **_kwargs):
        if args[:2] == ["get", "nodes"]:
            return nodes_payload
        raise AssertionError(f"unexpected kubectl json args: {args}")

    def fake_run_kubectl(_base_cmd, args, **_kwargs):
        if args[:4] == ["-n", "kube-system", "apply", "-f"]:
            return _completed(args, rc=1, err="image pull failed")
        return _completed(args)

    monkeypatch.setattr(topology_mod, "_kubectl_json", fake_kubectl_json)
    monkeypatch.setattr(topology_mod, "_run_kubectl", fake_run_kubectl)

    result = topology_mod.discover_cluster_topology(
        kubectl_base_cmd=["kubectl"],
        topology_root=tmp_path / "topology",
    )

    assert result.topology.collection_status == "failed"
    assert result.topology.discovered_nodes == 1
    node = result.topology.nodes[0]
    assert node.name == "pi"
    assert node.capabilities.monitoring_eligible is None
    assert "topology_unavailable" in node.capabilities.reasons
    assert (tmp_path / "topology" / "cluster.xml").is_file()