tianxuan/tests/test_e2e_v1.1.4.py

"""E2E test v1.1.4: upload 10 CSVs → cluster → verify globe.

Full pipeline WITHOUT manual server start/stop.
Port 8766 avoids conflicts with 8000.
Uses stdlib urllib (no pip install needed).
"""
import subprocess
import time
import sys
import os
import glob
import json
import uuid
import urllib.request
import urllib.error

PORT = 8766
BASE = f"http://127.0.0.1:{PORT}"
PROJ = r"C:\Users\25044\Desktop\Proj\天璇"


def http_get(path, timeout=30):
    """GET and return (status_code, body_bytes)."""
    try:
        r = urllib.request.urlopen(f"{BASE}{path}", timeout=timeout)
        body = r.read()
        code = r.status
        r.close()
        return code, body
    except urllib.error.HTTPError as e:
        return e.code, e.read()


def http_post_json(path, data_dict, timeout=30):
    """POST JSON and return (status_code, parsed_json)."""
    body = json.dumps(data_dict).encode("utf-8")
    req = urllib.request.Request(
        f"{BASE}{path}", data=body, headers={"Content-Type": "application/json"}
    )
    try:
        r = urllib.request.urlopen(req, timeout=timeout)
        return r.status, json.loads(r.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())


def http_post_files(path, file_paths, timeout=120):
    """POST multipart/form-data upload. Returns (status_code, parsed_json)."""
    boundary = uuid.uuid4().hex
    lines = []
    for fp in file_paths:
        fname = os.path.basename(fp)
        with open(fp, "rb") as f:
            filedata = f.read()
        lines.append(f"--{boundary}".encode())
        lines.append(
            f'Content-Disposition: form-data; name="files"; filename="{fname}"'.encode()
        )
        lines.append(b"Content-Type: text/csv")
        lines.append(b"")
        lines.append(filedata)
    lines.append(f"--{boundary}--".encode())
    body = b"\r\n".join(lines)

    req = urllib.request.Request(
        f"{BASE}{path}",
        data=body,
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
    )
    try:
        r = urllib.request.urlopen(req, timeout=timeout)
        return r.status, json.loads(r.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())


# ── 1. Start server ──
print("[1/6] Starting Django server...")
import tempfile as _tmp
_stderr_log = os.path.join(PROJ, "tests", "_server_stderr.log")
_stderr_fh = open(_stderr_log, "w")
server = subprocess.Popen(
    [
        os.path.join(PROJ, "runtime", "python", "python.exe"),
        "manage.py",
        "runserver",
        f"127.0.0.1:{PORT}",
        "--noreload",
    ],
    cwd=PROJ,
    stdout=_stderr_fh,
    stderr=subprocess.STDOUT,
)
time.sleep(8)

# Verify server is up
for attempt in range(10):
    try:
        urllib.request.urlopen(f"{BASE}/", timeout=3)
        print("  Server is up.")
        break
    except Exception:
        if attempt == 9:
            print("  FAIL: Server did not start within 30 seconds")
            server.terminate()
            server.wait()
            sys.exit(1)
        time.sleep(2)

# ── 2. Upload 10 test_*.csv files ──
csv_files = sorted(glob.glob(os.path.join(PROJ, "data", "multi_upload", "test_*.csv")))
print(f"[2/6] Found {len(csv_files)} CSV files. Uploading first 10...")
batch = csv_files[:10]
code, data = http_post_files("/upload/csv/", batch, timeout=120)

if code != 200:
    print(f"  FAIL: Upload returned HTTP {code}: {data}")
    server.terminate()
    server.wait()
    sys.exit(1)

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

# ── 3. Poll until upload processing completes (status='ready') ──
print("[3/6] Waiting for background processing...")
s = {}
for i in range(180):
    code, body = http_get(f"/runs/{run_id}/status/", timeout=10)
    s = json.loads(body) if code == 200 else {}
    if s.get("status") in ("ready", "failed"):
        break
    if i % 10 == 0:
        print(f"  Poll #{i}: {s.get('status')} ({s.get('progress_pct', 0)}%)")
    time.sleep(2)
print(f"  Status: {s.get('status')}, progress={s.get('progress_pct')}%")

if s.get("status") == "failed":
    print(f"  FAIL: {s.get('error_message', '')[:500]}")
    server.terminate()
    server.wait()
    sys.exit(1)

if s.get("status") != "ready":
    print(f"  FAIL: Timed out waiting for 'ready', got '{s.get('status')}'")
    server.terminate()
    server.wait()
    sys.exit(1)

# ── 4. Run clustering ──
print("[4/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.get('status')}")

if code != 200:
    print(f"  FAIL: Clustering start failed: {resp}")
    server.terminate()
    server.wait()
    sys.exit(1)

# ── 5. Poll until clustering completes ──
print("[5/6] Waiting for clustering...")
for i in range(120):
    code, body = http_get(f"/runs/{run_id}/status/", timeout=10)
    s = json.loads(body) if code == 200 else {}
    if s.get("status") in ("completed", "failed"):
        break
    if i % 5 == 0:
        print(
            f"  Poll #{i}: {s.get('status')} ({s.get('progress_pct', 0)}%)"
            f" - {s.get('progress_msg', '')}"
        )
    time.sleep(3)
print(
    f"  Final: status={s.get('status')}, entity_count={s.get('entity_count')},"
    f" cluster_count={s.get('cluster_count')}"
)

# ── 6. Verify globe page ──
print("[6/6] Verifying globe page...")
code, body = http_get(f"/globe/?runs={run_id}", timeout=30)
globe_text = body.decode("utf-8", errors="replace")
has_arcs = "arcCount" in globe_text
print(f"  Globe HTTP {code}, arcCount present: {has_arcs}")

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

# Print last 30 lines of server output on failure
_need_server_log = False

# ── Report ──
errors = []

if s.get("status") != "completed":
    errors.append(f"Expected status='completed', got '{s.get('status')}'")
    errors.append(f"error: {s.get('error_message', '')[:500]}")

if s.get("cluster_count", 0) == 0:
    errors.append("No clusters found (cluster_count=0)")

if not has_arcs:
    errors.append("Globe page missing 'arcCount' element")

if errors:
    print(f"\nFAIL ({len(errors)} errors):")
    for e in errors:
        print(f"  - {e}")
    # Show server stderr tail
    if os.path.exists(_stderr_log):
        with open(_stderr_log, "r", encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
            tail = lines[-30:] if len(lines) > 30 else lines
            print(f"\n--- Server stderr (last {len(tail)} lines) ---")
            for line in tail:
                print(f"  {line.rstrip()}")
    sys.exit(1)

print("\n*** E2E v1.1.4 PASSED ***")
print(
    f"  run_id={run_id}, clusters={s.get('cluster_count')},"
    f" entities={s.get('entity_count')}, globe_arcs={'yes' if has_arcs else 'no'}"
)