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