Nexus/tests/test_api.py

"""
Nexus API End-to-End Tests
Usage: python tests/test_api.py
Prerequisites: Python backend must be running (python -m uvicorn server.main:app)
"""
import json
import sys
import urllib.request
import urllib.error

BASE = "http://127.0.0.1:8600"
PASS = 0
FAIL = 0

# ── Auth (login first to get JWT) ──
ACCESS_TOKEN = ""
REFRESH_TOKEN = ""


def _get_auth_headers():
    return {"Authorization": f"Bearer {ACCESS_TOKEN}"} if ACCESS_TOKEN else {}


def test(name: str, method: str, path: str, body=None, expect_code=200, headers=None):
    """Execute an API test and report results."""
    global PASS, FAIL
    url = f"{BASE}{path}"
    data = json.dumps(body).encode() if body else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    # JWT auth
    for k, v in _get_auth_headers().items():
        req.add_header(k, v)
    # Extra headers
    if headers:
        for k, v in headers.items():
            req.add_header(k, v)
    try:
        resp = urllib.request.urlopen(req, timeout=10)
        code = resp.getcode()
        result = json.loads(resp.read())
        if code == expect_code:
            PASS += 1
            print(f"  [PASS] {name}")
            return result
        else:
            FAIL += 1
            print(f"  [FAIL] {name}: expected {expect_code}, got {code}")
    except urllib.error.HTTPError as e:
        code = e.code
        if code == expect_code:
            PASS += 1
            result_text = e.read().decode()[:200]
            try:
                return json.loads(result_text)
            except Exception:
                return {"detail": result_text}
        else:
            FAIL += 1
            print(f"  [FAIL] {name}: expected {expect_code}, got {code} — {e.read().decode()[:200]}")
    except Exception as e:
        FAIL += 1
        print(f"  [FAIL] {name}: {e}")
    return None


# ================================================================
print("=" * 50)
print("Nexus API End-to-End Tests")
print(f"  Target: {BASE}")
print("=" * 50)

# --- Health (no auth) ---
print("\n[1] Health")
test("GET /health", "GET", "/health")

# --- Auth: Login ---
print("\n[2] Auth")
login_result = test("POST /api/auth/login", "POST", "/api/auth/login", body={
    "username": "admin",
    "password": "admin",
}, expect_code=200)
if login_result and login_result.get("access_token"):
    ACCESS_TOKEN = login_result["access_token"]
    REFRESH_TOKEN = login_result.get("refresh_token", "")
    print(f"  → JWT token acquired (expires_in={login_result.get('expires_in')}s)")
else:
    print("  → WARNING: Login failed, subsequent tests will fail without JWT")

test("GET /api/auth/me", "GET", "/api/auth/me")

# --- Server CRUD ---
print("\n[3] Server CRUD")
created = test("POST /api/servers/ (create)", "POST", "/api/servers/", body={
    "name": "test-server-e2e",
    "domain": "192.168.1.100",
    "port": 22,
    "username": "root",
    "auth_method": "password",
    "password": "test123",
    "target_path": "/tmp/nexus-test",
    "category": "test",
})
server_id = created.get("id") if created and isinstance(created, dict) else 1

test("GET /api/servers/ (list)", "GET", "/api/servers/")
test("GET /api/servers/stats", "GET", "/api/servers/stats")
test(f"GET /api/servers/{server_id}", "GET", f"/api/servers/{server_id}")
test(f"PUT /api/servers/{server_id} (update)", "PUT", f"/api/servers/{server_id}", body={
    "description": "E2E test server",
})
test(f"DELETE /api/servers/{server_id}", "DELETE", f"/api/servers/{server_id}")

# --- Scripts ---
print("\n[4] Scripts")
script = test("POST /api/scripts/ (create)", "POST", "/api/scripts/", body={
    "name": "test-script",
    "category": "ops",
    "content": "echo hello",
})
script_id = script.get("id") if script and isinstance(script, dict) else 1
test("GET /api/scripts/ (list)", "GET", "/api/scripts/")
test(f"DELETE /api/scripts/{script_id}", "DELETE", f"/api/scripts/{script_id}")

# --- Schedules ---
print("\n[5] Schedules")
sched = test("POST /api/schedules/ (create)", "POST", "/api/schedules/", body={
    "name": "test-schedule",
    "source_path": "/www/wwwroot/",
    "cron_expr": "0 2 * * *",
    "server_ids": "[]",
    "enabled": True,
})
sched_id = sched.get("id") if sched and isinstance(sched, dict) else 1
test("GET /api/schedules/ (list)", "GET", "/api/schedules/")
test(f"PUT /api/schedules/{sched_id} (disable)", "PUT", f"/api/schedules/{sched_id}", body={
    "enabled": False,
})
test(f"DELETE /api/schedules/{sched_id}", "DELETE", f"/api/schedules/{sched_id}")

# --- Presets ---
print("\n[6] Password Presets")
preset = test("POST /api/presets/ (create)", "POST", "/api/presets/", body={
    "name": "test-preset",
    "encrypted_pw": "test-password-123",
})
preset_id = preset.get("id") if preset and isinstance(preset, dict) else 1
test("GET /api/presets/ (list)", "GET", "/api/presets/")
test(f"DELETE /api/presets/{preset_id}", "DELETE", f"/api/presets/{preset_id}")

# --- Settings ---
print("\n[7] Settings")
test("GET /api/settings/ (list)", "GET", "/api/settings/")
test("PUT /api/settings/system_name", "PUT", "/api/settings/system_name", body={
    "value": "Nexus",
})

# --- Audit ---
print("\n[8] Audit")
test("GET /api/audit/ (list)", "GET", "/api/audit/?limit=10")

# --- Retries ---
print("\n[9] Retry Queue")
test("GET /api/retries/ (list)", "GET", "/api/retries/")

# --- Sync ---
print("\n[10] Sync")
test("POST /api/sync/browse (no server)", "POST", "/api/sync/browse", body={
    "server_id": 99999,
    "path": "/tmp",
}, expect_code=404)

# ================================================================
total = PASS + FAIL
print(f"\n{'='*50}")
print(f"Results: {PASS}/{total} passed, {FAIL}/{total} failed")
if FAIL == 0:
    print("All tests passed!")
else:
    print(f"{FAIL} test(s) failed")

sys.exit(0 if FAIL == 0 else 1)