PM-pinou
133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""
|
|
Full-stack integration test. Run: python tests/test_integration.py
|
|
IMPORTANT: Update this file whenever frontend HTML/API changes.
|
|
Pure Python — no curl, no requests required.
|
|
"""
|
|
import subprocess, time, json, sys, threading, os, uuid, urllib.request, urllib.error
|
|
|
|
PORT = 18765
|
|
BASE = f"http://127.0.0.1:{PORT}"
|
|
errors = []
|
|
_lock = threading.Lock()
|
|
|
|
def fail(msg):
|
|
with _lock: errors.append(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',
|
|
)
|
|
def monitor():
|
|
for line in proc.stderr:
|
|
stripped = line.strip()
|
|
if 'ERROR' in stripped.upper() and 'GET' not in stripped and 'POST' not in stripped:
|
|
fail(f"STDERR: {stripped}")
|
|
threading.Thread(target=monitor, daemon=True).start()
|
|
return proc
|
|
|
|
def build_multipart_formdata(fields):
|
|
"""Build multipart/form-data body and content-type boundary.
|
|
fields: {name: (filename, content_type, data_bytes)}
|
|
"""
|
|
boundary = uuid.uuid4().hex
|
|
lines = []
|
|
for name, (filename, content_type, data) in fields.items():
|
|
lines.append(f'--{boundary}'.encode())
|
|
lines.append(f'Content-Disposition: form-data; name="{name}"; filename="{filename}"'.encode())
|
|
lines.append(f'Content-Type: {content_type}'.encode())
|
|
lines.append(b'')
|
|
lines.append(data)
|
|
lines.append(f'--{boundary}--'.encode())
|
|
return b'\r\n'.join(lines), f'multipart/form-data; boundary={boundary}'
|
|
|
|
def http_get(path, max_retries=10):
|
|
"""GET request with retry on connection refused."""
|
|
for attempt in range(max_retries):
|
|
try:
|
|
r = urllib.request.urlopen(f'{BASE}{path}', timeout=5)
|
|
body = r.read()
|
|
code = r.status
|
|
r.close()
|
|
return code, body
|
|
except urllib.error.URLError as e:
|
|
if 'refused' in str(e).lower() and attempt < max_retries - 1:
|
|
time.sleep(1)
|
|
continue
|
|
return 502, b'Connection refused'
|
|
|
|
def http_post_multipart(path, fields):
|
|
"""POST multipart form data, returns (status_code, body_bytes)."""
|
|
body_bytes, content_type = build_multipart_formdata(fields)
|
|
req = urllib.request.Request(f'{BASE}{path}', data=body_bytes,
|
|
headers={'Content-Type': content_type})
|
|
try:
|
|
r = urllib.request.urlopen(req, timeout=30)
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
|
|
def main():
|
|
proc = start_server()
|
|
time.sleep(3)
|
|
|
|
# Test 1: Homepage
|
|
code, _ = http_get('/')
|
|
check(code == 200, f"GET / → {code}")
|
|
|
|
# Test 2: Upload CSV with '+' in lat/lon
|
|
csv_content = (
|
|
":ips,:ipd,0ver,:prd,:prs,:ips.latd,:ips.lond\n"
|
|
"1.2.3.4,5.6.7.8,0303,443,80,35.0,139.0\n"
|
|
"9.10.11.12,13.14.15.16,0304,8080,443,+,+\n"
|
|
)
|
|
fields = {'files': ('test.csv', 'text/csv', csv_content.encode('utf-8'))}
|
|
code, body = http_post_multipart('/upload/csv/', fields)
|
|
check(code == 200, f"POST /upload/csv/ → {code}")
|
|
data = json.loads(body)
|
|
run_id = data.get('run_id')
|
|
check(run_id is not None, "No run_id in upload response")
|
|
|
|
# Test 3: Poll run status until ready
|
|
for _ in range(30):
|
|
code, body = http_get(f'/runs/{run_id}/status/')
|
|
check(code == 200, f"GET /runs/{run_id}/status/ → {code}")
|
|
s = json.loads(body)
|
|
if s['status'] in ('ready', 'completed', 'failed'):
|
|
check(s['status'] != 'failed', f"Run failed: {s.get('error_message','')}")
|
|
break
|
|
time.sleep(1)
|
|
|
|
# Test 4: Run list page
|
|
code, _ = http_get('/runs/')
|
|
check(code == 200, f"GET /runs/ → {code}")
|
|
|
|
# Test 5: Globe page
|
|
code, _ = http_get('/globe/')
|
|
check(code == 200, f"GET /globe/ → {code}")
|
|
|
|
# Test 6: Config page
|
|
code, _ = http_get('/config/')
|
|
check(code == 200, f"GET /config/ → {code}")
|
|
|
|
# Test 7: Delete run — @csrf_exempt means 200 (or 500), not 403, not 404
|
|
code, _ = http_post_multipart(f'/runs/{run_id}/delete/', {})
|
|
check(code != 404, f"POST /runs/{run_id}/delete/ → {code} (should not be 404)")
|
|
|
|
# Cleanup
|
|
proc.terminate()
|
|
proc.wait()
|
|
|
|
if errors:
|
|
print("INTEGRATION TEST FAILED:")
|
|
for e in errors: print(f" {e}")
|
|
sys.exit(1)
|
|
print("ALL INTEGRATION TESTS PASSED")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|