"""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