Nexus/tests/test_btpanel_ssh_bootstrap.py

"""Tests for Baota SSH bootstrap state machine and remote script parsing."""

import json

import pytest

from server.infrastructure.btpanel import ssh_bootstrap as ssh_bootstrap_module

from server.infrastructure.btpanel.bootstrap_state import (
    BOOTSTRAP_RETRY_SECONDS,
    STATE_FAILED,
    STATE_INSTALLING,
    STATE_PENDING,
    STATE_READY,
    apply_bootstrap_failure,
    apply_bootstrap_success,
    is_eligible_for_background_bootstrap,
    mark_bootstrap_pending,
    merge_bootstrap_fields,
)
from server.infrastructure.btpanel.ssh_bootstrap import (
    BootstrapRemoteError,
    build_bootstrap_command,
    classify_ssh_result,
    parse_bootstrap_stdout,
    ssh_bootstrap_panel,
    wrap_sudo_nopasswd,
)
from server.domain.models import Server


class _Srv:
    def __init__(self, extra_attrs=None):
        self.extra_attrs = extra_attrs or {}


def test_mark_bootstrap_pending_sets_now():
    extra = mark_bootstrap_pending({})
    block = extra["bt_panel"]
    assert block["bootstrap_state"] == STATE_PENDING
    assert block["bootstrap_next_attempt_at"] is not None
    assert block["auto_bootstrap"] is True


def test_apply_failure_bt_not_installed_goes_installing():
    extra = apply_bootstrap_failure({}, error_code="bt_not_installed", permanent=False)
    block = extra["bt_panel"]
    assert block["bootstrap_state"] == STATE_INSTALLING
    assert block["bootstrap_next_attempt_at"] is not None
    assert block["bootstrap_last_error"] == "bt_not_installed"


def test_apply_failure_ssh_auth_is_permanent():
    extra = apply_bootstrap_failure({}, error_code="ssh_auth_failed", permanent=True)
    block = extra["bt_panel"]
    assert block["bootstrap_state"] == STATE_FAILED
    assert block.get("bootstrap_next_attempt_at") is None


def test_apply_success_sets_ready(monkeypatch):
    monkeypatch.setattr(
        "server.infrastructure.btpanel.credentials.encrypt_value",
        lambda s: f"enc:{s}",
    )
    extra = apply_bootstrap_success(
        {},
        base_url="https://1.2.3.4:8888/abc",
        api_key="token123",
        verify_ssl=False,
        actions=["enabled_api"],
    )
    block = extra["bt_panel"]
    assert block["bootstrap_state"] == STATE_READY
    assert block["base_url"] == "https://1.2.3.4:8888/abc"
    assert block["api_key_enc"] == "enc:token123"
    assert block.get("bootstrap_next_attempt_at") is None


def test_is_eligible_respects_next_attempt():
    from datetime import datetime, timedelta, timezone

    past = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat().replace("+00:00", "Z")
    future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat().replace("+00:00", "Z")
    server_due = Server(
        id=1, name="a", domain="1.2.3.4",
        extra_attrs={"bt_panel": {
            "bootstrap_state": STATE_PENDING,
            "bootstrap_next_attempt_at": past,
            "auto_bootstrap": True,
        }},
    )
    server_later = Server(
        id=2, name="b", domain="1.2.3.5",
        extra_attrs={"bt_panel": {
            "bootstrap_state": STATE_INSTALLING,
            "bootstrap_next_attempt_at": future,
            "auto_bootstrap": True,
        }},
    )
    server_never_enrolled = Server(id=3, name="c", domain="1.2.3.6", extra_attrs={})
    assert is_eligible_for_background_bootstrap(server_due) is True
    assert is_eligible_for_background_bootstrap(server_later) is False
    assert is_eligible_for_background_bootstrap(server_never_enrolled) is False


def test_classify_ssh_auth_permanent():
    code, permanent = classify_ssh_result({
        "status": "success",
        "exit_code": 1,
        "stderr": "Permission denied (publickey).",
    })
    assert code == "ssh_auth_failed"
    assert permanent is True


def test_classify_api_json_permission_not_permanent():
    code, permanent = classify_ssh_result({
        "status": "success",
        "exit_code": 1,
        "stdout": "permission denied: '/www/server/panel/data/api.json'",
        "stderr": "",
    })
    assert code == "ssh_permission_denied"
    assert permanent is False


def test_parse_bootstrap_stdout_ok():
    payload = {"ok": True, "base_url": "https://x:8888", "api_key": "t", "actions": []}
    data = parse_bootstrap_stdout(json.dumps(payload))
    assert data["base_url"] == "https://x:8888"


def test_parse_bootstrap_stdout_remote_error():
    payload = {"ok": False, "error": "bt_not_installed", "message": "panel not installed"}
    with pytest.raises(BootstrapRemoteError) as exc:
        parse_bootstrap_stdout(json.dumps(payload))
    assert exc.value.code == "bt_not_installed"


def test_build_bootstrap_command_includes_payload():
    cmd = build_bootstrap_command(center_ip="10.0.0.1", panel_host="203.0.113.1")
    assert "10.0.0.1" in cmd
    assert "203.0.113.1" in cmd
    assert "mktemp /tmp/nexus_bt_bootstrap." in cmd
    assert "trap 'rm -f" in cmd
    assert "> /tmp/nexus_bt_bootstrap.py" not in cmd
    assert "python3 /tmp/nexus_bt_bootstrap.py" not in cmd


def test_remote_bootstrap_skips_reload_when_no_actions():
    from server.infrastructure.btpanel import ssh_bootstrap as mod

    script = mod._REMOTE_SCRIPT
    assert "if actions:" in script
    assert script.index("if actions:") < script.index('["/etc/init.d/bt", "reload"]')


def test_remote_bootstrap_patches_session_cleanup_ttl():
    from server.infrastructure.btpanel import ssh_bootstrap as mod

    script = mod._REMOTE_SCRIPT
    assert "patched_session_cleanup_ttl" in script
    assert "session_cleanup_timeout = int(public.get_session_timeout() or 86400)" in script
    assert "if f_time > session_cleanup_timeout" in script
    assert "hard_coded = \"                if f_time > 3600:\\n\"" in script
    assert "compile(patched, task_path, \"exec\")" in script
    assert "original_mode = os.stat(task_path).st_mode & 0o7777" in script
    assert "tempfile.mkstemp" in script
    assert "atomic_write_text(task_path, patched, original_mode)" in script
    assert "os.replace(tmp, path)" in script
    assert ".nexus-session-ttl.bak" in script
    assert "fcntl.flock" in script
    assert "LOCK_PATH" in script


def test_remote_bootstrap_uses_host_flock_and_unique_tmp():
    from server.infrastructure.btpanel import ssh_bootstrap as mod

    script = mod._REMOTE_SCRIPT
    assert "import fcntl" in script
    assert "fcntl.flock" in script
    assert "tempfile.mkstemp" in script
    assert 'task_path + ".nexus.tmp"' not in script
    assert 'API_PATH + ".nexus.tmp"' not in script


def test_wrap_sudo_nopasswd_root_unchanged():
    inner = "echo hi"
    assert wrap_sudo_nopasswd(inner, "root") == inner


def test_wrap_sudo_nopasswd_ubuntu():
    inner = "echo hi"
    wrapped = wrap_sudo_nopasswd(inner, "ubuntu")
    assert wrapped.startswith("sudo -n bash -c ")
    assert inner in wrapped


def test_classify_sudo_password_required():
    code, permanent = classify_ssh_result({
        "status": "success",
        "exit_code": 1,
        "stderr": "sudo: a password is required",
    })
    assert code == "ssh_sudo_required"
    assert permanent is True


def test_classify_ssh_result_success_exit_zero():
    ok_stdout = json.dumps(
        {"ok": True, "base_url": "https://x:8888", "api_key": "t", "actions": [], "verify_ssl": False}
    )
    code, permanent = classify_ssh_result(
        {"status": "success", "stdout": ok_stdout, "stderr": "", "exit_code": 0}
    )
    assert code == ""
    assert permanent is False


def test_retry_interval_constant():
    assert BOOTSTRAP_RETRY_SECONDS == 300



@pytest.mark.asyncio
async def test_ssh_bootstrap_panel_preserves_permanent_error_flag(monkeypatch):
    async def fake_exec_ssh_command(_server, _cmd, timeout=60, login_shell=False):
        return {
            "status": "success",
            "exit_code": 1,
            "stdout": "",
            "stderr": "sudo: a password is required",
        }

    monkeypatch.setattr(ssh_bootstrap_module, "exec_ssh_command", fake_exec_ssh_command)
    server = Server(id=9, name="sudo", domain="203.0.113.9", username="ubuntu", extra_attrs={})

    with pytest.raises(BootstrapRemoteError) as exc:
        await ssh_bootstrap_panel(server, center_ip="10.0.0.1")

    assert exc.value.code == "ssh_sudo_required"
    assert exc.value.permanent is True