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)

Env:
  NEXUS_TEST_BASE — default http://127.0.0.1:8600
  NEXUS_TEST_ADMIN_USER / NEXUS_TEST_ADMIN_PASSWORD — login credentials
"""
import json
import os
import sys
import urllib.request
import urllib.error

BASE = os.environ.get("NEXUS_TEST_BASE", "http://127.0.0.1:8600")
ADMIN_USER = os.environ.get("NEXUS_TEST_ADMIN_USER", "admin")
ADMIN_PASSWORD = os.environ.get("NEXUS_TEST_ADMIN_PASSWORD", "admin")


def _load_credentials_from_env_file():
    """Try to load test credentials from .env file (production deploy scenario).

    The gate check runs test_api.py automatically on the deploy server.
    Instead of hardcoding production passwords, read from .env variables:
      NEXUS_TEST_ADMIN_USER / NEXUS_TEST_ADMIN_PASSWORD
    These can be set in .env by the admin during installation.
    """
    env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env")
    if not os.path.exists(env_path):
        return
    try:
        with open(env_path) as f:
            for line in f:
                line = line.strip()
                if line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                key = key.strip()
                value = value.strip().strip("\"'")
                if key == "NEXUS_TEST_ADMIN_USER" and not os.environ.get("NEXUS_TEST_ADMIN_USER"):
                    os.environ["NEXUS_TEST_ADMIN_USER"] = value
                elif key == "NEXUS_TEST_ADMIN_PASSWORD" and not os.environ.get("NEXUS_TEST_ADMIN_PASSWORD"):
                    os.environ["NEXUS_TEST_ADMIN_PASSWORD"] = value
    except Exception:
        pass


# Load from .env before reading env vars
_load_credentials_from_env_file()
BASE = os.environ.get("NEXUS_TEST_BASE", "http://127.0.0.1:8600")
ADMIN_USER = os.environ.get("NEXUS_TEST_ADMIN_USER", "admin")
ADMIN_PASSWORD = os.environ.get("NEXUS_TEST_ADMIN_PASSWORD", "admin")
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 api_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()
        resp_body = resp.read()
        # Handle 204 No Content and empty responses
        if not resp_body or resp_body.strip() == b"":
            if code == expect_code:
                PASS += 1
                print(f"  [PASS] {name}")
                return {}
            else:
                FAIL += 1
                print(f"  [FAIL] {name}: expected {expect_code}, got {code}")
                return None
        result = json.loads(resp_body)
        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]
            if not result_text or result_text.strip() == "":
                print(f"  [PASS] {name}")
                return {}
            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


def api_test_plain_text(name: str, method: str, path: str, expect_body: str, expect_code=200):
    """Health and other probes that return plain text, not JSON."""
    global PASS, FAIL
    url = f"{BASE}{path}"
    req = urllib.request.Request(url, method=method)
    try:
        resp = urllib.request.urlopen(req, timeout=10)
        code = resp.getcode()
        body = resp.read().decode().strip()
        if code == expect_code and body == expect_body:
            PASS += 1
            print(f"  [PASS] {name}")
            return body
        FAIL += 1
        print(f"  [FAIL] {name}: expected {expect_code} body={expect_body!r}, got {code} body={body!r}")
    except urllib.error.HTTPError as e:
        FAIL += 1
        print(f"  [FAIL] {name}: HTTP {e.code} — {e.read().decode()[:200]}")
    except Exception as e:
        FAIL += 1
        print(f"  [FAIL] {name}: {e}")
    return None


def run_tests() -> int:
    """Run all API e2e tests. Returns exit code."""
    global PASS, FAIL, ACCESS_TOKEN, REFRESH_TOKEN
    PASS = 0
    FAIL = 0
    ACCESS_TOKEN = ""
    REFRESH_TOKEN = ""

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

    # --- Health (no auth) — plain text per health.py (Shell guardian) ---
    print("\n[1] Health")
    api_test_plain_text("GET /health", "GET", "/health", expect_body="ok")

    print("\n[2] Auth")
    login_result = api_test("POST /api/auth/login", "POST", "/api/auth/login", body={
        "username": ADMIN_USER,
        "password": ADMIN_PASSWORD,
    }, 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")

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

    print("\n[3] Server CRUD")
    existing_servers = api_test("GET /api/servers/ (pre-cleanup)", "GET", "/api/servers/?search=test-server-e2e", expect_code=200)
    if existing_servers and isinstance(existing_servers, dict):
        for srv in existing_servers.get("items", existing_servers.get("servers", [])):
            if isinstance(srv, dict) and srv.get("name") == "test-server-e2e":
                api_test("DELETE leftover test server", "DELETE", f"/api/servers/{srv['id']}", expect_code=200)
                break

    created = api_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",
    }, expect_code=201)
    server_id = created.get("id") if created and isinstance(created, dict) else None

    api_test("GET /api/servers/ (list)", "GET", "/api/servers/")
    api_test("GET /api/servers/stats", "GET", "/api/servers/stats")
    if server_id:
        api_test(f"GET /api/servers/{server_id}", "GET", f"/api/servers/{server_id}")
        api_test(f"PUT /api/servers/{server_id} (update)", "PUT", f"/api/servers/{server_id}", body={
            "description": "E2E test server",
        })
        api_test(f"DELETE /api/servers/{server_id}", "DELETE", f"/api/servers/{server_id}", expect_code=204)
    else:
        print("  → SKIP: Server CRUD detail tests (no server_id from create)")

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

    print("\n[5] Schedules")
    sched = api_test("POST /api/schedules/ (create)", "POST", "/api/schedules/", body={
        "name": "test-schedule",
        "source_path": "/tmp/nexus-test",
        "run_mode": "cron",
        "cron_expr": "0 2 * * *",
        "server_ids": "all",
        "enabled": True,
    }, expect_code=201)
    sched_id = sched.get("id") if sched and isinstance(sched, dict) else None
    api_test("GET /api/schedules/ (list)", "GET", "/api/schedules/")
    if sched_id:
        api_test(f"PUT /api/schedules/{sched_id} (disable)", "PUT", f"/api/schedules/{sched_id}", body={
            "enabled": False,
        })
        api_test(f"DELETE /api/schedules/{sched_id}", "DELETE", f"/api/schedules/{sched_id}", expect_code=204)

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

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

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

    print("\n[9] Retry Queue")
    api_test("GET /api/retries/ (list)", "GET", "/api/retries/")

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

    api_test("GET /api/files/browse (no server)", "GET", "/api/files/browse?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} api_test(s) failed")
    return 0 if FAIL == 0 else 1


if __name__ == "__main__":
    _load_credentials_from_env_file()
    sys.exit(run_tests())