test_endpoints.py

"""Quick endpoint tester - writes results to UTF-8 file."""
import urllib.request
import json
import time

BASE = "http://localhost:8000"
results = []

def log(msg):
    results.append(str(msg))

def test(name, url, method="GET", body=None):
    log(f"\n{'='*50}")
    log(f"TEST: {name}")
    log(f"URL: {url}")
    try:
        if body:
            data = json.dumps(body).encode()
            req = urllib.request.Request(url, data=data, method=method)
            req.add_header("Content-Type", "application/json")
        else:
            req = urllib.request.Request(url)
        start = time.time()
        resp = urllib.request.urlopen(req, timeout=60)
        elapsed = time.time() - start
        raw = resp.read().decode()
        d = json.loads(raw)
        log(f"STATUS: {resp.status} ({elapsed:.1f}s)")
        log(f"RESPONSE: {json.dumps(d, indent=2)[:2000]}")
        return d
    except Exception as e:
        log(f"ERROR: {e}")
        return None

# Test Root
test("Root", f"{BASE}/")

# Test Metrics
metrics = test("Metrics", f"{BASE}/api/metrics")
if metrics:
    resources = metrics.get("resources", [])
    log(f"Resources count: {len(resources)}")
    if resources:
        r = resources[0]
        log(f"First resource: {r['id']}, type: {r['type']}, metrics count: {len(r['metrics'])}")
        if r['metrics']:
            last = r['metrics'][-1]
            log(f"Latest metric: cpu={last.get('cpu')}, cost_usd={last.get('cost_usd')}, ts={last.get('timestamp')}")
    else:
        log("WARNING: resources array is EMPTY")

# Test Anomalies
anomalies = test("Anomalies", f"{BASE}/api/anomalies")
if anomalies:
    anoms = anomalies.get("anomalies", [])
    log(f"Anomalies count: {len(anoms)}")
    for a in anoms:
        log(f"  type={a.get('type')}, confidence={a.get('confidence')}, cost={a.get('cost_impact_usd')}")
        log(f"  claude_summary: {str(a.get('claude_summary',''))[:120]}")

# Test Remediate
result = test("Remediate", f"{BASE}/api/remediate", method="POST", body={
    "resource_id": "i-0da3659219976da09",
    "recommended_action": "stop_instance"
})

# Test Remediations list
test("Remediations List", f"{BASE}/api/remediations?limit=5")

# Test Forecast
log("\nTesting Forecast (may take 15-30s)...")
forecast = test("Forecast", f"{BASE}/api/forecast")
if forecast:
    fc = forecast.get("forecast", [])
    log(f"Forecast points: {len(fc)}")
    log(f"Spike warning: {forecast.get('spike_warning')}")
    log(f"Spike at: {forecast.get('spike_at')}")

# Check SQLite
log(f"\n{'='*50}")
log("SQLite data count:")
try:
    import sqlite3
    conn = sqlite3.connect("cloudseer.db")
    count = conn.execute("SELECT COUNT(*) FROM metrics").fetchone()[0]
    log(f"Metrics rows: {count}")
    if count < 10:
        log("WARNING: Less than 10 rows!")
    rem_count = conn.execute("SELECT COUNT(*) FROM remediations").fetchone()[0]
    log(f"Remediations rows: {rem_count}")
    conn.close()
except Exception as e:
    log(f"SQLite error: {e}")

log(f"\n{'='*50}")
log("ALL TESTS COMPLETE")

# Write results
with open("audit_results.txt", "w", encoding="utf-8") as f:
    f.write("\n".join(results))