Leonid PershinandCursor
2d404436bd
Expose /assistent subpaths for extension/DLL, overlay personas, roles, memory sqlite, live Assistent* API smoke, and optional chat_smoke so agents can diagnose missing tab, empty chat, and wrong models over HTTP. Co-authored-by: Cursor <cursoragent@cursor.com>
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""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
|
|
|
|
code, data = get("/assistent")
|
|
assert "playbook" in data
|
|
assert "hints" in data
|
|
assert data.get("error") == debug_checks.SSH_UNAVAILABLE or data.get("extension", {}).get(
|
|
"error"
|
|
) == debug_checks.SSH_UNAVAILABLE or not data.get("ok")
|
|
|
|
code, data = get("/assistent/api")
|
|
assert "ok" in data
|
|
|
|
code, data = get("/openapi.json")
|
|
assert "/assistent/extension" in data["paths"]
|
|
assert "/assistent/api" in data["paths"]
|
|
|
|
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_assistent_compact_and_roles_local(monkeypatch, tmp_path):
|
|
from gpu_rent import debug_assistent
|
|
|
|
compact = debug_assistent._compact_api(
|
|
"AssistentListModels",
|
|
{
|
|
"models": ["a:8b", "b:32b"],
|
|
"memory_models": ["nomic-embed-text"],
|
|
"preferred": "b:32b",
|
|
"base_url": "http://127.0.0.1:11434",
|
|
},
|
|
)
|
|
assert compact["model_count"] == 2
|
|
assert compact["preferred"] == "b:32b"
|
|
|
|
_auth_env(monkeypatch, tmp_path)
|
|
cfg = load_config(require_auth=True)
|
|
monkeypatch.setattr(
|
|
debug_assistent.debug_checks,
|
|
"load_state",
|
|
lambda: __import__("gpu_rent.state", fromlist=["SessionState"]).SessionState(
|
|
phase="idle"
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
debug_assistent,
|
|
"collect_assistent_fs",
|
|
lambda cfg, **k: {
|
|
"ok": True,
|
|
"extensions": [
|
|
{
|
|
"name": "swarm-assistent",
|
|
"path": "/mnt/swarm_data/Extensions/swarm-assistent",
|
|
"dll": "/x/SwarmAssistentExtension.dll",
|
|
"sqlite_dll": True,
|
|
"tab_html": True,
|
|
"bundle_js": True,
|
|
"git_head": "abc",
|
|
}
|
|
],
|
|
"overlay": {
|
|
"exists": True,
|
|
"persona_ids": ["leonid"],
|
|
"roles_present": True,
|
|
"roles": {
|
|
"chat": ["qwen3-vl:8b"],
|
|
"memory": ["nomic-embed-text"],
|
|
"default_chat": "qwen3-vl:8b",
|
|
},
|
|
"embed_model": "nomic-embed-text",
|
|
},
|
|
"sqlite": {"exists": True, "size": 100, "counts": {"memories": 2}},
|
|
"wanted": {"count": 0, "sample": []},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
debug_assistent.debug_checks,
|
|
"collect_ollama",
|
|
lambda cfg: {"ok": True, "models": ["qwen3-vl:8b", "nomic-embed-text"]},
|
|
)
|
|
monkeypatch.setattr(
|
|
debug_assistent,
|
|
"collect_assistent_api",
|
|
lambda cfg, chat_smoke=False: {
|
|
"ok": True,
|
|
"via": "local",
|
|
"calls": [
|
|
{"name": "AssistentListPersonas", "ok": True, "data": {"count": 3}},
|
|
{"name": "AssistentListModels", "ok": True, "data": {"model_count": 1}},
|
|
],
|
|
"hints": [],
|
|
},
|
|
)
|
|
deep = debug_assistent.collect_assistent_deep(cfg)
|
|
assert deep["ok"] is True
|
|
assert deep["extension"]["ok"] is True
|
|
assert deep["roles"]["roles"]["default_chat"] == "qwen3-vl:8b"
|
|
assert "playbook" in deep
|
|
|
|
# wrong default_chat
|
|
monkeypatch.setattr(
|
|
debug_assistent,
|
|
"collect_assistent_fs",
|
|
lambda cfg, **k: {
|
|
"ok": True,
|
|
"extensions": [
|
|
{
|
|
"name": "swarm-assistent",
|
|
"dll": "/x.dll",
|
|
"sqlite_dll": True,
|
|
"tab_html": True,
|
|
"bundle_js": True,
|
|
}
|
|
],
|
|
"overlay": {
|
|
"exists": True,
|
|
"persona_ids": [],
|
|
"roles_present": True,
|
|
"roles": {
|
|
"chat": ["missing:model"],
|
|
"memory": [],
|
|
"default_chat": "missing:model",
|
|
},
|
|
},
|
|
"sqlite": {"exists": False, "size": 0},
|
|
"wanted": {"count": 0, "sample": []},
|
|
},
|
|
)
|
|
roles = debug_assistent.collect_assistent_roles(cfg)
|
|
assert roles["ok"] is False
|
|
assert any("default_chat" in h for h in roles["hints"])
|
|
|
|
|
|
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
|