gpu-rent/tests/test_debug_api.py

"""Unit tests for localhost Debug API sidecar."""

from __future__ import annotations

import json
import time
import urllib.request
from dataclasses import replace
from pathlib import Path

import pytest

from gpu_rent import debug_api, debug_checks
from gpu_rent.config import load_config
from gpu_rent.state import SessionState


def _auth_env(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("USERPROFILE", str(tmp_path))
    monkeypatch.chdir(tmp_path)
    for key, val in {
        "OS_AUTH_URL": "https://example/v3",
        "OS_USER_DOMAIN_NAME": "default",
        "OS_USERNAME": "u",
        "OS_PASSWORD": "secret-password",
        "OS_PROJECT_ID": "proj",
        "OS_REGION_NAME": "ru-7",
        "GPU_RENT_AZ": "ru-7a",
        "CIVITAI_API_TOKEN": "civ-secret",
        "HF_TOKEN": "hf-secret",
        "GIT_TOKEN": "git-secret",
        "SELECTEL_API_TOKEN": "sel-secret",
        "DEBUG_LOCAL_PORT": "0",  # overridden per-test with free port
    }.items():
        monkeypatch.setenv(key, val)


def test_redact_config_hides_secrets(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    cfg = load_config(require_auth=True)
    red = debug_checks.redact_config(cfg)
    assert red["os_password"] == "***"
    assert red["civitai_api_token"] == "***"
    assert red["hf_token"] == "***"
    assert red["git_token"] == "***"
    assert red["selectel_api_token"] == "***"
    assert red["os_username"] == "u"
    assert cfg.os_password == "secret-password"  # original untouched


def test_redact_state_notes(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    st = SessionState(phase="provisioning", floating_ip="1.2.3.4")
    st.notes = {"idle_killer": "armed", "api_token": "leak"}
    red = debug_checks.redact_state(st)
    assert red["phase"] == "provisioning"
    assert red["notes"]["api_token"] == "***"
    assert red["notes"]["idle_killer"] == "armed"


def test_ssh_unavailable_without_fip(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    cfg = load_config(require_auth=True)
    monkeypatch.setattr(debug_checks, "load_state", lambda: SessionState(phase="idle"))
    out = debug_checks.collect_logs(cfg)
    assert out["ok"] is False
    assert out["error"] == debug_checks.SSH_UNAVAILABLE
    assert out["phase"] == "idle"


def test_debug_server_routes(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    cfg = load_config(require_auth=True)
    # Free ephemeral port
    import socket

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    cfg = replace(cfg, debug_local_port=port)

    logs: list[str] = []
    srv = debug_api.start_debug_server(cfg, port=port, log=logs.append, print_urls=True)
    assert srv is not None
    assert any("openapi.json" in x for x in logs)
    try:
        tee = srv.wrap_log(lambda m: None)
        tee("жду ready backend…")
        srv.set_step("wait_backend_idle")
        time.sleep(0.15)

        def get(path: str) -> tuple[int, dict | str]:
            with urllib.request.urlopen(f"http://127.0.0.1:{port}{path}", timeout=5) as resp:
                raw = resp.read().decode("utf-8")
                code = getattr(resp, "status", 200)
                if path == "/":
                    return code, raw
                return code, json.loads(raw)

        code, body = get("/")
        assert code == 200
        assert "snapshot" in body

        code, data = get("/openapi.json")
        assert code == 200
        assert "/health" in data["paths"]

        code, data = get("/progress")
        assert data["ok"] is True
        assert data["step"] in {"wait_backend_idle", "starting"} or "backend" in data["step"]

        code, data = get("/events?since=0")
        assert data["ok"] is True
        assert any("ready backend" in e["msg"] for e in data["events"])

        code, data = get("/config")
        assert data["config"]["os_password"] == "***"

        code, data = get("/state")
        assert "state" in data

        code, data = get("/snapshot")
        assert "phase" in data
        assert "checks" in data

        code, data = get("/health")
        assert "uptime_sec" in data

        code, data = get("/logs")
        assert data["error"] == debug_checks.SSH_UNAVAILABLE

        with pytest.raises(Exception):
            urllib.request.urlopen(f"http://127.0.0.1:{port}/nope", timeout=2)
    finally:
        debug_api.stop_debug_server(srv)


def test_bind_fail_returns_none(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    cfg = load_config(require_auth=True)
    import socket

    blocker = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    blocker.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    blocker.bind(("127.0.0.1", 0))
    port = blocker.getsockname()[1]
    # Keep bound so second bind fails on Windows/Linux
    try:
        warns: list[str] = []
        # On some platforms SO_REUSEADDR allows double-bind; force fail via invalid host trick
        # by patching ThreadingHTTPServer
        class Boom:
            def __init__(self, *a, **k):
                raise OSError("Address already in use")

        monkeypatch.setattr(debug_api, "ThreadingHTTPServer", Boom)
        srv = debug_api.start_debug_server(cfg, port=port, log=warns.append)
        assert srv is None
        assert any("не стартовал" in w for w in warns)
    finally:
        blocker.close()


def test_config_debug_port_default(monkeypatch, tmp_path):
    _auth_env(monkeypatch, tmp_path)
    monkeypatch.delenv("DEBUG_LOCAL_PORT", raising=False)
    cfg = load_config(require_auth=True)
    assert cfg.debug_local_port == 17821