neuron-tai/tests/test_mining_cli.py

"""Tests for US-016: mining-style node startup CLI + live dashboard."""

from __future__ import annotations

import json
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock, patch


# ---------------------------------------------------------------------------
# model_catalog tests
# ---------------------------------------------------------------------------

def test_curated_models_list_is_non_empty():
    from meshnet_node.model_catalog import CURATED_MODELS
    assert len(CURATED_MODELS) >= 5


def test_model_preset_vram_for_quant():
    from meshnet_node.model_catalog import CURATED_MODELS

    m = CURATED_MODELS[0]  # Llama-3-70B
    assert m.vram_for_quant("nf4") == m.vram_nf4
    assert m.vram_for_quant("int8") == m.vram_int8
    assert m.vram_for_quant("bf16") == m.vram_bf16
    assert m.vram_for_quant("bfloat16") == m.vram_bf16  # alias


def test_model_preset_fits_vram():
    from meshnet_node.model_catalog import CURATED_MODELS

    small = next(m for m in CURATED_MODELS if m.vram_nf4 < 10)
    assert small.fits_vram(small.vram_nf4, "nf4")
    assert not small.fits_vram(small.vram_nf4 - 1, "nf4")


def test_recommended_quant_respects_vram():
    from meshnet_node.model_catalog import CURATED_MODELS

    m = CURATED_MODELS[0]  # Llama-3-70B: nf4=18, int8=40, bf16=140
    assert m.recommended_quant(200) == "bf16"
    assert m.recommended_quant(50) == "int8"
    assert m.recommended_quant(20) == "nf4"
    assert m.recommended_quant(5) is None


def test_models_with_insufficient_vram_are_marked(monkeypatch):
    from meshnet_node import wizard as wiz

    # Simulate 6 GB GPU
    gpus = [{"index": 0, "name": "RTX 3060", "vram_gb": 6.0, "backend": "cuda"}]
    monkeypatch.setattr(wiz, "_detect_gpus", lambda: gpus)

    # Phi-3 at NF4 needs 4 GB — should fit; Llama-3-70B at NF4 needs 18 GB — should not
    from meshnet_node.model_catalog import CURATED_MODELS

    phi = next(m for m in CURATED_MODELS if "Phi-3" in m.name)
    llama = next(m for m in CURATED_MODELS if "Llama-3-70B" in m.name)

    assert phi.fits_vram(6.0, "nf4")
    assert not llama.fits_vram(6.0, "nf4")


# ---------------------------------------------------------------------------
# config tests
# ---------------------------------------------------------------------------

def test_load_config_returns_none_when_missing(tmp_path):
    from meshnet_node.config import load_config
    assert load_config(tmp_path / "nonexistent.json") is None


def test_save_and_load_config_roundtrip(tmp_path):
    from meshnet_node.config import save_config, load_config

    cfg = {"model_hf_repo": "test/model", "quantization": "nf4", "tracker_url": "http://localhost:8080"}
    cfg_path = tmp_path / "config.json"
    save_config(cfg, cfg_path)

    loaded = load_config(cfg_path)
    assert loaded == cfg


def test_save_config_creates_parent_dirs(tmp_path):
    from meshnet_node.config import save_config, load_config

    nested = tmp_path / "deep" / "nested" / "config.json"
    save_config({"x": 1}, nested)
    assert nested.exists()
    assert load_config(nested) == {"x": 1}


def test_merge_cli_overrides_applies_non_none_values():
    from meshnet_node.config import merge_cli_overrides

    base = {"tracker_url": "http://a:8080", "quantization": "nf4", "port": 7000}
    result = merge_cli_overrides(base, tracker_url="http://b:9090", port=None)
    assert result["tracker_url"] == "http://b:9090"
    assert result["port"] == 7000  # None override ignored
    assert result["quantization"] == "nf4"  # unchanged


# ---------------------------------------------------------------------------
# wizard tests
# ---------------------------------------------------------------------------

def test_print_models_table_runs_without_error(capsys, monkeypatch):
    from meshnet_node import wizard as wiz

    monkeypatch.setattr(wiz, "_detect_gpus", lambda: [{"index": 0, "name": "GPU", "vram_gb": 24.0, "backend": "cuda"}])
    wiz.print_models_table()
    out = capsys.readouterr().out
    assert "Llama" in out or "Qwen" in out or "Phi" in out


def test_wizard_writes_config_on_happy_path(tmp_path, monkeypatch):
    from meshnet_node import wizard as wiz
    from meshnet_node.config import load_config, save_config

    # Fake GPU
    gpus = [{"index": 0, "name": "RTX 4090", "vram_gb": 24.0, "backend": "cuda"}]
    monkeypatch.setattr(wiz, "_detect_gpus", lambda: gpus)
    # Tracker not reachable (stub)
    monkeypatch.setattr(wiz, "_ping_tracker", lambda url: False)

    # Simulate user selecting model 3 (Mixtral), quant 1 (nf4), default dir, default tracker, default wallet
    inputs = iter([
        "3",          # pick Mixtral (index 3 in CURATED_MODELS)
        "1",          # quant NF4
        str(tmp_path / "models"),  # download dir
        "http://localhost:8080",   # tracker
        str(tmp_path / "wallet.json"),  # wallet
    ])
    monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs))

    cfg = wiz.run_wizard(config_path_override=tmp_path / "config.json")
    assert cfg["model_hf_repo"] == "mistralai/Mixtral-8x7B-Instruct-v0.1"
    assert cfg["quantization"] == "nf4"
    assert "download_dir" in cfg
    assert cfg["tracker_url"] == "http://localhost:8080"


def test_wizard_raises_keyboard_interrupt_on_ctrl_c(monkeypatch):
    from meshnet_node import wizard as wiz

    gpus = [{"index": 0, "name": "RTX 4090", "vram_gb": 24.0, "backend": "cuda"}]
    monkeypatch.setattr(wiz, "_detect_gpus", lambda: gpus)

    call_count = [0]

    def fake_input(prompt=""):
        call_count[0] += 1
        if call_count[0] == 1:
            raise KeyboardInterrupt

    monkeypatch.setattr("builtins.input", fake_input)

    import pytest
    with pytest.raises(KeyboardInterrupt):
        wiz.run_wizard()


# ---------------------------------------------------------------------------
# dashboard tests
# ---------------------------------------------------------------------------

def test_is_interactive_tty_false_when_not_tty(monkeypatch):
    from meshnet_node import dashboard as dash

    monkeypatch.setattr(sys.stdout, "isatty", lambda: False)
    assert not dash.is_interactive_tty()


def test_dashboard_plain_fallback_on_keyboard_interrupt(monkeypatch):
    """Plain loop exits cleanly when Ctrl-C is raised."""
    from meshnet_node import dashboard as dash

    node = MagicMock()
    node.chat_completion_count = 5

    call_count = [0]

    def fake_sleep(t):
        call_count[0] += 1
        if call_count[0] >= 1:
            raise KeyboardInterrupt

    monkeypatch.setattr(dash.time, "sleep", fake_sleep)
    monkeypatch.setattr(dash, "_gpu_stats", lambda: [])
    monkeypatch.setattr(sys.stdout, "isatty", lambda: False)

    cfg = {"model_name": "test-model", "quantization": "nf4"}
    # Should not raise
    dash.run_dashboard(node, cfg, start_time=dash.time.monotonic())


def test_ema_updates_correctly():
    from meshnet_node.dashboard import _EMA

    ema = _EMA(alpha=1.0)  # alpha=1.0 → always takes latest sample
    ema.update(10.0)
    assert ema.value == 10.0
    ema.update(20.0)
    assert ema.value == 20.0


# ---------------------------------------------------------------------------
# CLI integration tests
# ---------------------------------------------------------------------------

def test_models_command_prints_table(capsys, monkeypatch):
    """meshnet-node models prints the curated table and exits 0."""
    from meshnet_node import wizard as wiz

    monkeypatch.setattr(wiz, "_detect_gpus", lambda: [])

    from meshnet_node.cli import main
    monkeypatch.setattr(sys, "argv", ["meshnet-node", "models"])

    try:
        main()
    except SystemExit as exc:
        assert exc.code == 0

    out = capsys.readouterr().out
    assert "Llama" in out or "Qwen" in out or "Phi" in out


def test_config_command_no_config_exits_1(tmp_path, monkeypatch):
    from meshnet_node import config as cfg_mod
    from meshnet_node.cli import main

    monkeypatch.setattr(cfg_mod, "_DEFAULT_CONFIG_FILE", tmp_path / "nonexistent.json")
    monkeypatch.setattr(sys, "argv", ["meshnet-node", "config"])

    with patch("meshnet_node.config.config_path", return_value=tmp_path / "nonexistent.json"):
        try:
            main()
        except SystemExit as exc:
            assert exc.code == 1


def test_config_command_prints_saved_config(tmp_path, monkeypatch, capsys):
    from meshnet_node import config as cfg_mod
    from meshnet_node.config import save_config
    from meshnet_node.cli import main

    saved = {"model_hf_repo": "meta-llama/Meta-Llama-3-70B-Instruct", "quantization": "nf4"}
    cfg_file = tmp_path / "config.json"
    save_config(saved, cfg_file)

    monkeypatch.setattr(sys, "argv", ["meshnet-node", "config"])

    with patch("meshnet_node.config.config_path", return_value=cfg_file):
        with patch("meshnet_node.config.load_config", return_value=saved):
            try:
                main()
            except SystemExit as exc:
                assert exc.code == 0

    out = capsys.readouterr().out
    data = json.loads(out.split("\n", 1)[1])  # skip the "Config: ..." header line
    assert data["model_hf_repo"] == saved["model_hf_repo"]


def test_legacy_start_subcommand_accepted(monkeypatch):
    """meshnet-node start --tracker http://... does not crash on arg parsing."""
    from meshnet_node.cli import main

    def fake_run_startup(*args, **kwargs):
        class _FakeNode:
            chat_completion_count = 0
            def stop(self): pass
        return _FakeNode()

    monkeypatch.setattr(sys, "argv", [
        "meshnet-node", "start",
        "--tracker", "http://localhost:8080",
        "--model", "stub-model",
        "--port", "0",
    ])

    raised = []

    def fake_sleep(t):
        raise KeyboardInterrupt

    with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
        with patch("time.sleep", side_effect=fake_sleep):
            try:
                main()
            except SystemExit as exc:
                raised.append(exc.code)

    # Exited (either 0 or via KeyboardInterrupt caught in _cmd_start)
    # The important thing is no unhandled exception from arg parsing