tianxuan/tests/test_e2e_v1.1.6.py

"""E2E test v1.1.6: generate 50 CSVs → upload 10 → cluster → verify globe.

Self-contained: generates data, starts server, uploads, clusters, verifies globe, reports.
Port 18766. Uses requests for HTTP (installed automatically if missing).
"""
import subprocess
import time
import sys
import os
import glob
import json
import urllib.request
import urllib.error

PORT = 18766
BASE = f"http://127.0.0.1:{PORT}"
PROJ = r"C:\Users\25044\Desktop\Proj\天璇"
PYTHON = os.path.join(PROJ, "runtime", "python", "python.exe")

# ── Ensure requests is installed ──
try:
    import requests
except ImportError:
    print("[0] Installing requests...")
    subprocess.run([PYTHON, "-m", "pip", "install", "requests", "-q"], cwd=PROJ, check=True, timeout=120)
    import requests


def http_get(path, timeout=30):
    """GET and return (status_code, parsed_json or body_bytes)."""
    try:
        r = requests.get(f"{BASE}{path}", timeout=timeout)
        try:
            return r.status_code, r.json()
        except Exception:
            return r.status_code, r.text
    except Exception as e:
        return 0, str(e)


def http_post_json(path, data_dict, timeout=30):
    """POST JSON and return (status_code, parsed_json)."""
    try:
        r = requests.post(f"{BASE}{path}", json=data_dict, timeout=timeout)
        try:
            return r.status_code, r.json()
        except Exception:
            return r.status_code, r.text
    except Exception as e:
        return 0, str(e)


def http_post_files(path, file_paths, timeout=120):
    """POST multipart/form-data upload. Returns (status_code, parsed_json)."""
    files = [("files", (os.path.basename(fp), open(fp, "rb"), "text/csv")) for fp in file_paths]
    try:
        r = requests.post(f"{BASE}{path}", files=files, timeout=timeout)
        return r.status_code, r.json()
    except Exception as e:
        return 0, str(e)


def poll_status(run_id, target_status, max_wait_sec, description):
    """Poll /runs/<run_id>/status/ until target_status or timeout. Returns status dict."""
    s = {}
    for i in range(max_wait_sec // 2):
        code, body = http_get(f"/runs/{run_id}/status/", timeout=10)
        s = body if isinstance(body, dict) else {}
        if s.get("status") in (target_status, "failed"):
            break
        if i % 10 == 0:
            print(f"  [{description}] Poll #{i}: {s.get('status')} ({s.get('progress_pct', 0)}%)")
        time.sleep(2)
    print(f"  [{description}] Final: status={s.get('status')}, progress={s.get('progress_pct', 0)}%")
    return s


# ═══════════════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════════════

errors = []

# ── 1. Generate 50 test CSV files ──
print("[1/6] Generating 50 test CSV files (100 rows each)...")
gen_result = subprocess.run(
    [PYTHON, "scripts/gen_multi.py", "--files", "50", "--rows", "100"],
    cwd=PROJ, capture_output=True, text=True, timeout=180,
)
if gen_result.returncode != 0:
    print(f"  FAIL: gen_multi.py failed.\nSTDERR:\n{gen_result.stderr}")
    sys.exit(1)
print(gen_result.stdout.strip())

csv_files = sorted(glob.glob(os.path.join(PROJ, "data", "multi_upload", "test_*.csv")))
print(f"  Generated {len(csv_files)} CSV files.")

# ── 2. Start Django server ──
print(f"[2/6] Starting Django server on port {PORT}...")
stderr_log = os.path.join(PROJ, "tests", "_e2e_server_v116.log")
stderr_fh = open(stderr_log, "w")
server = subprocess.Popen(
    [PYTHON, "manage.py", "runserver", f"127.0.0.1:{PORT}", "--noreload"],
    cwd=PROJ, stdout=stderr_fh, stderr=subprocess.STDOUT,
)
time.sleep(6)

# Verify server is up
server_up = False
for attempt in range(15):
    try:
        urllib.request.urlopen(f"{BASE}/", timeout=3)
        server_up = True
        print("  Server is up.")
        break
    except Exception:
        if attempt == 14:
            errors.append("Server did not start within 45 seconds")
        time.sleep(2)

if not server_up:
    server.terminate(); server.wait(); stderr_fh.close()
    print("\nFAIL")
    for e in errors: print(f"  - {e}")
    sys.exit(1)

# ── 3. Upload first 10 CSV files ──
print("[3/6] Uploading first 10 CSV files...")
batch = csv_files[:10]
code, data = http_post_files("/upload/csv/", batch, timeout=180)

if code != 200:
    errors.append(f"Upload returned HTTP {code}: {str(data)[:300]}")
    server.terminate(); server.wait(); stderr_fh.close()
    print("\nFAIL")
    for e in errors: print(f"  - {e}")
    sys.exit(1)

run_id = data["run_id"]
print(f"  Uploaded {len(batch)} files → run_id={run_id}, status={data['status']}")

# ── 4. Poll until background processing completes (status='ready') ──
print("[4/6] Waiting for background upload processing (max 300s)...")
s = poll_status(run_id, "ready", 300, "Upload processing")

if s.get("status") == "failed":
    errors.append(f"Upload processing failed: {s.get('error_message', '')[:500]}")

if s.get("status") not in ("ready",):
    errors.append(f"Expected status='ready', got '{s.get('status')}'")

if errors:
    server.terminate(); server.wait(); stderr_fh.close()
    print("\nFAIL")
    for e in errors: print(f"  - {e}")
    sys.exit(1)

# ── 5. Run clustering and poll ──
print("[5/6] Starting clustering (HDBSCAN, min_cluster_size=5)...")
code, resp = http_post_json(
    "/analyze/run/",
    {"run_id": run_id, "algorithm": "hdbscan", "min_cluster_size": 5, "cluster_mode": "raw"},
    timeout=30,
)
print(f"  Response HTTP {code}: {resp if isinstance(resp, dict) else str(resp)[:200]}")

if code != 200:
    errors.append(f"Clustering start failed: {resp}")
    server.terminate(); server.wait(); stderr_fh.close()
    print("\nFAIL")
    for e in errors: print(f"  - {e}")
    sys.exit(1)

s_cluster = poll_status(run_id, "completed", 600, "Clustering")

if s_cluster.get("status") == "failed":
    errors.append(f"Clustering failed: {s_cluster.get('error_message', '')[:500]}")

if s_cluster.get("status") != "completed":
    errors.append(f"Expected status='completed', got '{s_cluster.get('status')}'")

# ── 6. Verify globe page ──
print("[6/6] Verifying globe page...")
code, globe_body = http_get(f"/globe/?runs={run_id}", timeout=30)
globe_text = globe_body if isinstance(globe_body, str) else json.dumps(globe_body)
has_arcs = "arcCount" in globe_text if isinstance(globe_text, str) else False
print(f"  Globe HTTP {code}, arcCount present: {has_arcs}")

# ── Cleanup ──
server.terminate()
server.wait()
stderr_fh.close()

# ── Collect metrics ──
total_flows = s_cluster.get("total_flows", "?")
entity_count = s_cluster.get("entity_count", 0)
cluster_count = s_cluster.get("cluster_count", 0)
status = s_cluster.get("status", "unknown")

# ── Validate required metrics ──
if not has_arcs:
    errors.append("Globe page missing 'arcCount' element")
if entity_count == 0:
    errors.append("entity_count=0 (no entities detected)")
if cluster_count == 0:
    errors.append("cluster_count=0 (no clusters found)")

# ── Show server log tail on failure ──
if errors and os.path.exists(stderr_log):
    with open(stderr_log, "r", encoding="utf-8", errors="replace") as f:
        lines = f.readlines()
        tail = lines[-40:] if len(lines) > 40 else lines
        print(f"\n--- Server log tail ({len(tail)} lines) ---")
        for line in tail:
            print(f"  {line.rstrip()}")

# ── Report ──
print()
print("=" * 60)
print("  E2E v1.1.6 RESULTS")
print("=" * 60)
print(f"  run_id        : {run_id}")
print(f"  status        : {status}")
print(f"  total_flows   : {total_flows}")
print(f"  entity_count  : {entity_count}")
print(f"  cluster_count : {cluster_count}")
print(f"  globe_arcs    : {'PASS' if has_arcs else 'FAIL'}")

if errors:
    print(f"\n  FAIL ({len(errors)} errors):")
    for e in errors:
        print(f"    - {e}")
    sys.exit(1)
else:
    print(f"\n  *** E2E v1.1.6 PASSED ***")
    print(f"  run_id={run_id}, clusters={cluster_count}, entities={entity_count}, globe_arcs=yes")
    sys.exit(0)