bootnode/api/tests/test_lux_fleet_overview.py

"""Tests for the read-only Lux fleet overview collector (lux_fleet.py).

Pure-logic + mocked-IO coverage:
  - CR parsing (lux.cloud/v1 shape): version, plugin-set, validators, phase.
  - Pod-proxy addressing: the one way the collector reaches a validator.
  - RPC enrichment (mocked httpx) → version/peers/heights/bootstrapped, and a
    regression guard that every luxd call goes to /v1 (luxd serves no /ext).
  - Overview aggregation + totals.
  - Graceful "no lux-k8s credential" shaped-empty response.
"""

from unittest.mock import patch

import httpx
import pytest

from bootnode.core.chains import lux_fleet
from bootnode.core.chains.lux_fleet import (
    ChainHeight,
    DexSummary,
    LuxNode,
    _parse_cr,
    _pod_proxy,
    get_overview,
)


def _dex_stub() -> DexSummary:
    return DexSummary(d_chain_bootstrapped=False, note="D-Chain not yet bootstrapped")


# --- Sample data mirroring the live devnet LuxNetwork CR (lux.cloud/v1) -----

SAMPLE_CR = {
    "apiVersion": "lux.cloud/v1",
    "kind": "LuxNetwork",
    "spec": {
        "validators": 5,
        "image": {"repository": "ghcr.io/luxfi/node", "tag": "v1.36.25"},
        "pluginSource": {
            "bucket": "s3://lux-plugins-devnet/v1.3.1/",
            "vmPlugins": [
                {"objectKey": "dexvm-linux-amd64", "vmID": "dexvm"},
                {"objectKey": "evm-linux-amd64", "vmID": "mgj786NP"},
                {"objectKey": "no-id"},  # missing vmID is skipped
            ],
        },
    },
    "status": {"phase": "Unknown", "observedGeneration": 27},
}


class TestParseCR:
    def test_extracts_declared_shape(self):
        p = _parse_cr(SAMPLE_CR)
        assert p["validators"] == 5
        assert p["image"] == "ghcr.io/luxfi/node:v1.36.25"
        assert p["version"] == "v1.36.25"
        assert p["plugin_set"] == "s3://lux-plugins-devnet/v1.3.1/"
        assert p["phase"] == "Unknown"

    def test_vm_plugins_skip_missing_ids(self):
        p = _parse_cr(SAMPLE_CR)
        assert p["vm_plugins"] == ["dexvm", "mgj786NP"]

    def test_empty_cr_safe(self):
        p = _parse_cr({})
        assert p["validators"] == 0
        assert p["phase"] == "Unknown"
        assert p["version"] is None
        assert p["plugin_set"] is None
        assert p["vm_plugins"] == []


class TestPodProxy:
    def test_addresses_the_pod_through_the_api_server(self):
        assert _pod_proxy("lux-mainnet", "luxd-3", 9630) == (
            "/api/v1/namespaces/lux-mainnet/pods/luxd-3:9630/proxy"
        )

    def test_port_is_per_network(self):
        ports = {n: c["http_port"] for n, c in lux_fleet.LUX_NETWORKS.items()}
        assert ports == {"mainnet": 9630, "testnet": 9640, "devnet": 9650}


class TestCollectNetwork:
    @pytest.mark.asyncio
    async def test_cr_and_pods_merged_with_rpc(self):
        """CR declares the shape, pods observe it, RPC fills in live state.
        luxd-0/1/2 answer RPC; luxd-3/4 are up but their RPC does not."""
        pods = {i: {"phase": "Running", "ready": True, "ip": f"10.0.0.{i}"}
                for i in range(5)}

        async def fake_probe(http, base, node: LuxNode):
            assert base == _pod_proxy("lux-devnet", node.pod_name, 9650)
            if node.pod_index > 2:
                return
            node.rpc_reachable = True
            node.version = "luxd/1.36.25"
            node.node_id = f"NodeID-{node.pod_index}"
            node.peers = 4
            node.healthy = True
            node.chains = [
                ChainHeight(chain="C", bootstrapped=True, height=108),
                ChainHeight(chain="P", bootstrapped=True, height=0),
                ChainHeight(chain="X", bootstrapped=True),
                ChainHeight(chain="D", bootstrapped=False),
            ]

        with patch.object(lux_fleet, "_read_luxnetwork_cr", return_value=SAMPLE_CR), \
             patch.object(lux_fleet, "_list_luxd_pods", return_value=pods), \
             patch.object(lux_fleet, "_probe_node_rpc", side_effect=fake_probe), \
             patch.object(lux_fleet, "_dex_summary", side_effect=lambda n, net: _dex_stub()):
            fleet = await lux_fleet._collect_network(
                "devnet", custom_api=object(), core_api=object(), http=object())

        assert fleet.network == "devnet"
        assert fleet.namespace == "lux-devnet"
        assert fleet.version == "luxd/1.36.25"      # observed, not the CR tag
        assert fleet.image == "ghcr.io/luxfi/node:v1.36.25"   # declared
        assert fleet.plugin_set == "s3://lux-plugins-devnet/v1.3.1/"
        assert fleet.validators == 5
        assert fleet.nodes_total == 5
        assert fleet.nodes_ready == 5               # pod readiness
        assert fleet.nodes_rpc_ok == 3              # only 3 answer RPC
        assert fleet.total_peers == 12              # 3 nodes * 4 peers
        assert fleet.dex.d_chain_bootstrapped is False
        assert [n.ip for n in fleet.nodes] == [f"10.0.0.{i}" for i in range(5)]
        n3 = next(n for n in fleet.nodes if n.pod_index == 3)
        assert n3.rpc_reachable is False
        assert n3.phase == "Running"

    @pytest.mark.asyncio
    async def test_version_derived_from_rpc_when_cr_absent(self):
        pods = {0: {"phase": "Running", "ready": True, "ip": "10.0.0.9"}}

        async def fake_probe(http, base, node: LuxNode):
            node.rpc_reachable = True
            node.version = "luxd/1.36.2"
            node.peers = 7
            node.chains = [ChainHeight(chain="D", bootstrapped=True)]

        with patch.object(lux_fleet, "_read_luxnetwork_cr", return_value={}), \
             patch.object(lux_fleet, "_list_luxd_pods", return_value=pods), \
             patch.object(lux_fleet, "_probe_node_rpc", side_effect=fake_probe), \
             patch.object(lux_fleet, "_dex_summary", side_effect=lambda n, net: _dex_stub()):
            fleet = await lux_fleet._collect_network(
                "mainnet", custom_api=object(), core_api=object(), http=object())

        assert fleet.nodes_total == 1
        assert fleet.validators == 1              # derived from node count
        assert fleet.version == "luxd/1.36.2"     # derived from RPC
        assert fleet.phase == "Running"
        assert fleet.total_peers == 7

    @pytest.mark.asyncio
    async def test_running_version_wins_over_a_stale_cr(self):
        """lux-operator is scaled to 0, so a CR tag can be arbitrarily stale.
        Mainnet's CR says v1.34.0 — a tag that does not exist — while all five
        nodes run luxd/1.36.2. The dashboard must show what is running."""
        pods = {i: {"phase": "Running", "ready": True, "ip": f"10.2.0.{i}"}
                for i in range(5)}
        stale_cr = {"spec": {"validators": 5,
                             "image": {"repository": "ghcr.io/luxfi/node",
                                       "tag": "v1.34.0"}},
                    "status": {"phase": "Running"}}

        async def fake_probe(http, base, node: LuxNode):
            node.rpc_reachable = True
            node.version = "luxd/1.36.2"

        with patch.object(lux_fleet, "_read_luxnetwork_cr", return_value=stale_cr), \
             patch.object(lux_fleet, "_list_luxd_pods", return_value=pods), \
             patch.object(lux_fleet, "_probe_node_rpc", side_effect=fake_probe), \
             patch.object(lux_fleet, "_dex_summary", side_effect=lambda n, net: _dex_stub()):
            fleet = await lux_fleet._collect_network(
                "mainnet", custom_api=object(), core_api=object(), http=object())

        assert fleet.version == "luxd/1.36.2"
        assert fleet.image == "ghcr.io/luxfi/node:v1.34.0"

    @pytest.mark.asyncio
    async def test_unreachable_cluster_reports_error_not_silence(self):
        def boom(*a, **kw):
            raise RuntimeError("connection refused")

        with patch.object(lux_fleet, "_read_luxnetwork_cr", side_effect=boom), \
             patch.object(lux_fleet, "_list_luxd_pods", side_effect=boom):
            fleet = await lux_fleet._collect_network(
                "testnet", custom_api=object(), core_api=object(), http=object())

        assert fleet.error is not None
        assert "connection refused" in fleet.error
        assert fleet.nodes_total == 0


class TestOverview:
    @pytest.mark.asyncio
    async def test_missing_credential_yields_shaped_error(self):
        def boom():
            raise RuntimeError("no kubeconfig")

        with patch.object(lux_fleet, "_load_lux_k8s_clients", side_effect=boom):
            ov = await get_overview()

        assert len(ov.networks) == 3
        assert all(f.error and "no kubeconfig" in f.error for f in ov.networks)
        assert ov.totals["nodes_rpc_ok"] == 0

    @pytest.mark.asyncio
    async def test_totals_aggregate_every_network(self):
        pods = {i: {"phase": "Running", "ready": True, "ip": f"10.1.0.{i}"}
                for i in range(5)}

        async def fake_probe(http, base, node: LuxNode):
            node.rpc_reachable = True
            node.version = "luxd/1.36.2"
            node.peers = 3

        with patch.object(lux_fleet, "_load_lux_k8s_clients",
                          return_value=(object(), object(), {"base_url": "https://k8s"})), \
             patch.object(lux_fleet, "_read_luxnetwork_cr", return_value=SAMPLE_CR), \
             patch.object(lux_fleet, "_list_luxd_pods", return_value=pods), \
             patch.object(lux_fleet, "_probe_node_rpc", side_effect=fake_probe), \
             patch.object(lux_fleet, "_dex_summary", side_effect=lambda n, net: _dex_stub()):
            ov = await get_overview()

        assert ov.totals == {
            "networks": 3, "validators": 15, "nodes_total": 15,
            "nodes_ready": 15, "nodes_rpc_ok": 15, "total_peers": 45,
        }


class TestRpcParsing:
    @pytest.mark.asyncio
    async def test_probe_parses_rpc_results_and_only_calls_v1(self):
        """_probe_node_rpc maps luxd JSON-RPC into the node model.

        Also asserts every request path is under the pod proxy's /v1 — luxd
        serves no /ext, so an /ext call is a silent all-null dashboard.
        """
        base = _pod_proxy("lux-devnet", "luxd-0", 9650)
        seen: list[str] = []

        def router(request: httpx.Request) -> httpx.Response:
            seen.append(request.url.path)
            if request.url.path.endswith("/v1/health"):
                return httpx.Response(200, json={"healthy": True})
            body = request.read().decode()
            if "info.getNodeVersion" in body:
                return httpx.Response(200, json={"result": {
                    "version": "luxd/1.36.25", "gitCommit": "abc123",
                    "databaseVersion": "v1.4.5"}})
            if "info.getNodeID" in body:
                return httpx.Response(200, json={"result": {"nodeID": "NodeID-XYZ"}})
            if "info.peers" in body:
                return httpx.Response(200, json={"result": {"peers": [1, 2, 3, 4]}})
            if "info.isBootstrapped" in body:
                ch = "C" if '"C"' in body else ("D" if '"D"' in body else "?")
                return httpx.Response(200, json={"result": {"isBootstrapped": ch != "D"}})
            if "eth_blockNumber" in body:
                return httpx.Response(200, json={"result": "0x6c"})
            if "platform.getHeight" in body:
                return httpx.Response(200, json={"result": {"height": "42"}})
            return httpx.Response(404)

        node = LuxNode(pod_name="luxd-0", pod_index=0, ip="10.0.0.1")
        async with httpx.AsyncClient(base_url="https://k8s",
                                     transport=httpx.MockTransport(router)) as http:
            await lux_fleet._probe_node_rpc(http, base, node)

        assert node.rpc_reachable is True
        assert node.version == "luxd/1.36.25"
        assert node.git_commit == "abc123"
        assert node.node_id == "NodeID-XYZ"
        assert node.peers == 4
        assert node.healthy is True
        heights = {c.chain: c for c in node.chains}
        assert heights["C"].height == 108
        assert heights["P"].height == 42
        assert heights["C"].bootstrapped is True
        assert heights["D"].bootstrapped is False

        assert seen, "no requests were made"
        assert all(p.startswith(f"{base}/v1/") for p in seen), seen
        assert not any("/ext" in p for p in seen), seen

    @pytest.mark.asyncio
    async def test_degraded_node_503_health_is_data_not_unknown(self):
        """luxd answers /v1/health with 503 + a body when degraded."""
        base = _pod_proxy("lux-mainnet", "luxd-2", 9630)

        def router(request: httpx.Request) -> httpx.Response:
            return httpx.Response(503, json={"healthy": False})

        async with httpx.AsyncClient(base_url="https://k8s",
                                     transport=httpx.MockTransport(router)) as http:
            assert await lux_fleet._node_health(http, base) is False