tianxuan/tests/test_e2e_full.py

"""
Full-stack end-to-end test: upload → process → analyze workflow.

1. Start Django server on port 18766, monitor stderr
2. Upload ALL CSV files from data/multi_upload/ via POST /upload/csv/
3. Poll /runs/{id}/status/ until 'ready' or 'failed'
4. POST /analyze/llm/ to run LLM auto-analysis (real DeepSeek LLM)
5. Poll /runs/{id}/status/ until 'completed' or 'failed'
6. Check status is 'completed', not 'failed'
7. GET /globe/ → 200
8. POST /runs/{id}/delete/ → 200
9. Stop server
10. FAIL on any error → sys.exit(1)

Usage:
  runtime/python/python.exe tests/test_e2e_full.py            # Full test with LLM
  runtime/python/python.exe tests/test_e2e_full.py --skip-llm  # Upload+process only, skip LLM
"""
import subprocess
import time
import json
import sys
import threading
import os
import uuid
import argparse
import urllib.request
import urllib.error
import glob

PORT = 18766
BASE = f"http://127.0.0.1:{PORT}"
FAILED = []
_lock = threading.Lock()


def fail(msg):
    with _lock:
        FAILED.append(msg)
    print(f"  FAIL: {msg}")


def check(cond, msg):
    if not cond:
        fail(msg)


def start_server():
    manage_py = os.path.join(os.path.dirname(__file__), '..', 'manage.py')
    proc = subprocess.Popen(
        [sys.executable, manage_py, 'runserver', f'127.0.0.1:{PORT}', '--noreload'],
        stderr=subprocess.PIPE, stdout=subprocess.PIPE,
        text=True, encoding='utf-8', errors='replace',
    )

    def monitor():
        for line in proc.stderr:
            stripped = line.strip()
            upper = stripped.upper()
            # Django dev server logs request lines like "GET /... 200" to stderr — skip those
            # Also skip LLM tool diagnostics (tool errors are expected and handled)
            if ('ERROR' in upper and 'GET' not in upper and 'POST' not in upper
                    and 'tianxuan.llm' not in stripped
                    and 'this error occurred' not in stripped.lower()
                    and 'traceback' not in stripped.lower()):
                fail(f"STDERR: {stripped}")

    threading.Thread(target=monitor, daemon=True).start()

    # Wait for server to be ready
    for _ in range(15):
        try:
            urllib.request.urlopen(f'{BASE}/', timeout=3)
            return proc
        except Exception:
            time.sleep(1)
    raise RuntimeError("Server did not start")


def http_get(path):
    r = urllib.request.urlopen(f'{BASE}{path}', timeout=15)
    body = r.read()
    code = r.status
    r.close()
    return code, body


def http_post_json(path, data_dict):
    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=30)
        return r.status, json.loads(r.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())


def http_post_raw(path, data=None):
    """POST with optional raw data (no content-type override)."""
    req = urllib.request.Request(f'{BASE}{path}', data=data or b'')
    try:
        r = urllib.request.urlopen(req, timeout=30)
        body = r.read()
        code = r.status
        r.close()
        return code, body
    except urllib.error.HTTPError as e:
        return e.code, e.read()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--skip-llm', action='store_true',
                        help='Skip LLM analysis steps (4-6), only test upload+processing')
    args = parser.parse_args()

    proc = start_server()
    print("[1/8] Server started")

    # ── Upload ALL CSV files in a single request ──
    data_dir = os.path.join(os.path.dirname(__file__), '..', 'data', 'multi_upload')
    csv_files = sorted(glob.glob(os.path.join(data_dir, '*.csv')))
    print(f"[2/8] Uploading {len(csv_files)} CSV files (single multipart request)...")

    boundary = uuid.uuid4().hex
    lines = []
    for fp in csv_files:
        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)
    ct = f'multipart/form-data; boundary={boundary}'

    req = urllib.request.Request(f'{BASE}/upload/csv/', data=body,
                                 headers={'Content-Type': ct})
    try:
        r = urllib.request.urlopen(req, timeout=300)
        resp = json.loads(r.read())
        r.close()
    except urllib.error.HTTPError as e:
        err_body = e.read().decode('utf-8', errors='replace')
        fail(f"Upload failed: HTTP {e.code} - {err_body[:300]}")
        proc.terminate()
        proc.wait()
        sys.exit(1)
    except Exception as e:
        fail(f"Upload exception: {e}")
        proc.terminate()
        proc.wait()
        sys.exit(1)

    run_id = resp.get('run_id')
    check(run_id is not None, "No run_id in upload response")
    total_files = resp.get('total_files', len(csv_files))
    total_rows = resp.get('total_rows', '?')
    print(f"  Uploaded {total_files} files ({total_rows} rows), Run ID: {run_id}")

    # ── Poll status until 'ready' ──
    print("[3/8] Waiting for processing (background: load -> profile -> ready)...")
    ready = False
    final_status_info = {}
    for i in range(180):
        code, resp_body = http_get(f'/runs/{run_id}/status/')
        check(code == 200, f"Status poll -> {code}")
        s = json.loads(resp_body)
        status = s['status']
        final_status_info = s
        if status == 'failed':
            err = s.get('error_message', '')[:200]
            fail(f"Run failed: {err}")
            break
        if status == 'ready':
            ready = True
            print(f"  Status: {status} ({s.get('progress_pct', 0)}%)")
            break
        if i % 10 == 0:
            print(f"  Poll #{i}: {status} ({s.get('progress_pct', 0)}%)")
        time.sleep(2)

    check(ready, "Run never became ready after upload")
    if not ready:
        proc.terminate()
        proc.wait()
        sys.exit(1)

    # ── Report upload stats ──
    run_total_files = final_status_info.get('total_files', total_files)
    run_total_rows = final_status_info.get('total_rows', total_rows)
    print(f"\n  === UPLOAD RESULTS ===")
    print(f"  Status:        {final_status_info.get('status')}")
    print(f"  Progress:      {final_status_info.get('progress_pct', 0)}%")
    print(f"  Total files:   {run_total_files}")
    print(f"  Total rows:    {run_total_rows}")
    csv_glob_info = final_status_info.get('csv_glob', '')
    if csv_glob_info:
        print(f"  CSV glob:      {csv_glob_info}")
    print(f"  ====================\n")

    if args.skip_llm:
        print("[4/8] SKIP: LLM analysis (--skip-llm)")
        print("[5/8] SKIP: LLM polling (--skip-llm)")
        print("[6/8] Verifying upload status (--skip-llm mode)...")
        check(status == 'ready', f"Run status '{status}' after upload, expected 'ready'")
        print(f"  Upload status: {status} [OK]")
    else:
        # ── POST /analyze/llm/ to run real LLM auto-analysis ──
        print("[4/8] Starting LLM auto-analysis (real DeepSeek LLM, may take 2-5 min)...")
        code, resp = http_post_json('/analyze/llm/', {'run_id': run_id})
        check(code == 200, f"LLM start -> HTTP {code}")
        llm_status = resp.get('status')
        print(f"  LLM started: {llm_status}")

        # ── Poll status until 'completed' or 'failed' (LLM phase) ──
        print("[5/8] Waiting for LLM analysis to complete...")
        completed = False
        for i in range(120):  # 120 × 3s = 360s = 6 min max
            code, resp_body = http_get(f'/runs/{run_id}/status/')
            check(code == 200, f"LLM status poll -> {code}")
            s = json.loads(resp_body)
            status = s['status']
            pct = s.get('progress_pct', 0)
            msg = s.get('progress_msg', '')
            if status == 'failed':
                err = s.get('error_message', '') or s.get('progress_msg', '') or '(no details)'
                fail(f"LLM analysis failed: {err[:300]}")
                break
            if status == 'completed':
                completed = True
                print(f"  LLM completed ({pct}%)")
                run_log = s.get('run_log', '')
                if run_log:
                    log_snippet = run_log[-200:].encode('ascii', errors='replace').decode('ascii')
                    print(f"  Last log: {log_snippet}")
                break
            if i % 5 == 0 or i == 10:
                print(f"  Poll #{i}: {status} ({pct}%) - {msg}")
            time.sleep(3)

        check(completed, "LLM analysis never completed")
        if not completed:
            proc.terminate()
            proc.wait()
            sys.exit(1)

        # ── Check status is 'completed', not 'failed' ──
        print("[6/8] Verifying final status...")
        code, resp_body = http_get(f'/runs/{run_id}/status/')
        s = json.loads(resp_body) if code == 200 else {}
        final_status = s.get('status', 'unknown')
        check(final_status == 'completed', f"Final status is '{final_status}', expected 'completed'")
        print(f"  Final status: {final_status} [OK]")

    # ── GET /globe/ → 200 ──
    print("[7/8] Checking globe page...")
    code, _ = http_get('/globe/')
    check(code == 200, f"GET /globe/ -> {code}")
    print(f"  Globe -> HTTP {code} [OK]")

    # ── POST /runs/{run_id}/delete/ → 200 ──
    print("[8/8] Deleting run...")
    code, _ = http_post_raw(f'/runs/{run_id}/delete/')
    check(code == 200, f"DELETE -> HTTP {code}")
    print(f"  Delete -> HTTP {code} [OK]")

    # ── Cleanup ──
    proc.terminate()
    proc.wait()

    if FAILED:
        print(f"\n!!! INTEGRATION TEST FAILED: {len(FAILED)} failures")
        for e in FAILED:
            print(f"  {e}")
        sys.exit(1)
    print("\n*** ALL INTEGRATION TESTS PASSED ***")


if __name__ == '__main__':
    main()