prole/tests/installer/test_topology.py

from __future__ import annotations

import json
import subprocess

from knoe.core import topology as topology_mod
from knoe.core import storage_probe as storage_probe_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)
        if args[:4] == ["-n", "kube-system", "delete", "pod"]:
            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,
            storage_probe_enabled=False,
        ),
    )

    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",
        config=topology_mod.TopologyDiscoveryConfig(storage_probe_enabled=False),
    )

    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()


def test_discover_cluster_topology_includes_storage_items(tmp_path, monkeypatch) -> None:
    nodes_payload = {
        "items": [
            {
                "metadata": {
                    "name": "arthur",
                    "labels": {"kubernetes.io/arch": "amd64", "kubernetes.io/os": "linux"},
                },
                "spec": {},
                "status": {
                    "conditions": [{"type": "Ready", "status": "True"}],
                    "addresses": [{"type": "InternalIP", "address": "10.0.0.20"}],
                    "allocatable": {"cpu": "4", "memory": "8Gi", "ephemeral-storage": "50Gi"},
                },
            }
        ]
    }

    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="collector disabled in test")
        return _completed(args)

    def fake_probe(node, **_kwargs):
        item = storage_probe_mod.StorageProbeResult(
            target_id="abc123",
            node_name=node.name,
            label="Synology d004",
            mountpoint="/synology/d004",
            source_type="synology_mount",
            storage_class="synology",
            succeeded=True,
            skipped=False,
            error=None,
            metrics=storage_probe_mod.StorageProbeMetrics(
                randread_iops=620.0,
                randwrite_iops=240.0,
                read_bw_bytes=2_539_520,
                write_bw_bytes=983_040,
                read_lat_ms_p95=18.0,
                write_lat_ms_p95=24.7,
                test_runtime_sec=8.1,
            ),
            io_class="io_slow_durable",
            timing_multiplier=3.0,
            observed_at="2026-03-26T00:00:00+00:00",
        )
        return storage_probe_mod.NodeStorageInventory(node_name=node.name, items=[item])

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

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

    assert result.topology.discovered_nodes == 1
    node = result.topology.nodes[0]
    assert len(node.storage_items) == 1
    assert node.storage_items[0].label == "Synology d004"
    assert node.storage_items[0].io_class == "io_slow_durable"


def test_collector_fs_stats_permission_denied_does_not_abort_collection() -> None:
    script = topology_mod._collector_python_script()
    namespace: dict[str, object] = {}
    exec(script, namespace)

    class _PathWithDeniedExists:
        def __init__(self, value: str) -> None:
            self.value = value

        def __truediv__(self, other: str):
            base = self.value.rstrip("/")
            child = str(other).lstrip("/")
            return _PathWithDeniedExists(f"{base}/{child}" if base else f"/{child}")

        def exists(self) -> bool:
            raise PermissionError("denied")

        def __str__(self) -> str:
            return self.value

    class _Stat:
        f_frsize = 4096
        f_blocks = 1000
        f_bavail = 200
        f_bfree = 300

    def fake_statvfs(path: str):
        if "run/user/1000/doc" in path:
            raise PermissionError("denied")
        return _Stat()

    namespace["Path"] = _PathWithDeniedExists
    namespace["_mount_rows"] = lambda: [
        ("/dev/sda1", "/run/user/1000/doc", "ext4"),
        ("/dev/sdb1", "/data", "ext4"),
    ]
    namespace["_read_rotational"] = lambda _block_name: None
    namespace["_read_transport"] = lambda *_args: "unknown"
    namespace["_infer_storage_class"] = lambda *_args: "unknown"
    namespace["os"].statvfs = fake_statvfs

    filesystems = namespace["_collect_filesystems"]()

    assert len(filesystems) == 2
    first = filesystems[0]
    second = filesystems[1]
    assert first["mountpoint"] == "/run/user/1000/doc"
    assert first["sizeBytes"] is None
    assert first["usedBytes"] is None
    assert first["availableBytes"] is None
    assert second["mountpoint"] == "/data"
    assert second["sizeBytes"] == 4096 * 1000
    assert second["availableBytes"] == 4096 * 200