226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
"""上传解析 e2e：上传 → 解析回填全链路（覆盖 md/txt/pdf/html/xlsx/pptx/URL + 异常）。
|
|
|
|
用法：python tests/e2e_upload.py
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import uuid
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
BASE = "http://127.0.0.1:5099"
|
|
USER = f"up_e2e_{int(time.time())}"
|
|
PASS = "up_e2e_123"
|
|
TMP = ROOT / "tests" / "_up"
|
|
|
|
COOKIE = {"value": ""}
|
|
|
|
|
|
def call(method, path, payload=None, raw=False):
|
|
url = BASE + path
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
if COOKIE["value"]:
|
|
req.add_header("Cookie", COOKIE["value"])
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
body = resp.read()
|
|
sc = resp.headers.get("Set-Cookie")
|
|
if sc and not COOKIE["value"]:
|
|
COOKIE["value"] = sc.split(";")[0]
|
|
return resp.status, (body if raw else json.loads(body or b"{}"))
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read()
|
|
try:
|
|
parsed = json.loads(body or b"{}")
|
|
except Exception:
|
|
parsed = {"raw": body.decode("utf-8", "ignore")}
|
|
return e.code, parsed
|
|
|
|
|
|
def upload(filename, data, with_cookie=True):
|
|
"""multipart 上传，返回 (status, body)。"""
|
|
boundary = "----pptagente2e" + uuid.uuid4().hex[:8]
|
|
head = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; '
|
|
f'filename="{filename}"\r\nContent-Type: application/octet-stream\r\n\r\n')
|
|
body = head.encode("utf-8") + data + f"\r\n--{boundary}--\r\n".encode("utf-8")
|
|
req = urllib.request.Request(BASE + "/api/upload", data=body, method="POST")
|
|
req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")
|
|
if with_cookie and COOKIE["value"]:
|
|
req.add_header("Cookie", COOKIE["value"])
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
return resp.status, json.loads(resp.read() or b"{}")
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
return e.code, json.loads(e.read() or b"{}")
|
|
except Exception:
|
|
return e.code, {}
|
|
|
|
|
|
def make_samples():
|
|
"""生成本地样例文件，返回 {格式名: (文件名, 字节)}。"""
|
|
TMP.mkdir(exist_ok=True)
|
|
samples = {}
|
|
|
|
md = "# UPLOAD_E2E_MARKER_MD\n知识库平台建设背景说明\n本期完成检索服务上线\n"
|
|
samples["md"] = ("sample.md", md.encode("utf-8"))
|
|
|
|
txt = "UPLOAD_E2E_MARKER_TXT\n知识库平台建设背景说明\n本期完成检索服务上线"
|
|
samples["txt"] = ("sample.txt", txt.encode("utf-8"))
|
|
|
|
html = ("<html><head><style>.x{color:red}</style></head><body>"
|
|
"<h1>UPLOAD_E2E_MARKER_HTML</h1><p>知识库平台季度汇报</p></body></html>")
|
|
samples["html"] = ("sample.html", html.encode("utf-8"))
|
|
|
|
import fitz
|
|
doc = fitz.open()
|
|
page = doc.new_page()
|
|
page.insert_text((72, 72), "UPLOAD_E2E_MARKER_PDF quarterly report of knowledge platform")
|
|
buf = io.BytesIO()
|
|
doc.save(buf)
|
|
samples["pdf"] = ("sample.pdf", buf.getvalue())
|
|
|
|
from openpyxl import Workbook
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws["A1"] = "UPLOAD_E2E_MARKER_XLSX"
|
|
ws["A2"] = "knowledge platform kpi"
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
samples["xlsx"] = ("sample.xlsx", buf.getvalue())
|
|
|
|
from pptx import Presentation
|
|
from pptx.util import Inches
|
|
prs = Presentation()
|
|
slide = prs.slides.add_slide(prs.slide_layouts[5])
|
|
tb = slide.shapes.add_textbox(Inches(1), Inches(1), Inches(6), Inches(1))
|
|
tb.text_frame.text = "UPLOAD_E2E_MARKER_PPTX knowledge platform review"
|
|
buf = io.BytesIO()
|
|
prs.save(buf)
|
|
samples["pptx"] = ("sample.pptx", buf.getvalue())
|
|
|
|
return samples
|
|
|
|
|
|
def start_url_server():
|
|
"""本地起一个极简网页服务器，供 URL 抓取分支测试。"""
|
|
class H(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
body = (b"<html><head><title>t</title></head><body>"
|
|
b"<h1>UPLOAD_E2E_MARKER_URL</h1><p>web capture ok</p></body></html>")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
srv = HTTPServer(("127.0.0.1", 5199), H)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
return srv
|
|
|
|
|
|
def main():
|
|
failures = []
|
|
|
|
def check(name, cond, extra=""):
|
|
print(f" [{'OK' if cond else 'NG'}] {name} {extra}")
|
|
if not cond:
|
|
failures.append(name)
|
|
|
|
env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
|
|
subprocess.run([sys.executable, "-m", "src.db.init"], cwd=str(ROOT),
|
|
capture_output=True, env=env)
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-c",
|
|
f"import sys; sys.path.insert(0, r'{ROOT}');"
|
|
"from src.web.app import app; app.run(host='127.0.0.1', port=5099)"],
|
|
cwd=str(ROOT), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
srv = start_url_server()
|
|
samples = make_samples()
|
|
try:
|
|
up = False
|
|
for _ in range(30):
|
|
time.sleep(0.5)
|
|
try:
|
|
call("GET", "/api/me")
|
|
up = True
|
|
break
|
|
except Exception:
|
|
continue
|
|
check("server up", up)
|
|
|
|
s, b = call("POST", "/api/register",
|
|
{"username": USER, "password": PASS, "confirm_password": PASS})
|
|
check("register", s == 200 and b.get("code") == 0)
|
|
|
|
# ---- 鉴权与非法输入 ----
|
|
s, b = upload("x.md", b"# hi", with_cookie=False)
|
|
check("upload without login -> 401", s == 401, str(s))
|
|
s, b = upload("evil.exe", b"MZ fake")
|
|
check("upload exe -> 3001", s == 400 and b.get("code") == 3001, str(b)[:60])
|
|
s, b = upload("huge.txt", b"A" * (21 * 1024 * 1024))
|
|
check("upload 21MB -> 3002", b.get("code") == 3002, str(b)[:60])
|
|
s, b = call("POST", "/api/parse", {"file_id": "nothex"})
|
|
check("parse bad file_id -> 400", s == 400, str(b)[:60])
|
|
s, b = call("POST", "/api/parse", {"file_id": "f" * 32})
|
|
check("parse unknown file_id -> 404", s == 404 and b.get("code") == 3003, str(b)[:60])
|
|
|
|
# ---- 各格式解析 ----
|
|
file_ids = {}
|
|
for fmt, (fname, data) in samples.items():
|
|
s, b = upload(fname, data)
|
|
ok_up = s == 200 and b.get("code") == 0 and b.get("data", {}).get("file_id")
|
|
check(f"upload {fmt}", bool(ok_up), str(b)[:60])
|
|
if not ok_up:
|
|
continue
|
|
fid = b["data"]["file_id"]
|
|
file_ids[fmt] = fid
|
|
marker = {
|
|
"md": "UPLOAD_E2E_MARKER_MD", "txt": "UPLOAD_E2E_MARKER_TXT",
|
|
"html": "UPLOAD_E2E_MARKER_HTML", "pdf": "UPLOAD_E2E_MARKER_PDF",
|
|
"xlsx": "UPLOAD_E2E_MARKER_XLSX", "pptx": "UPLOAD_E2E_MARKER_PPTX",
|
|
}[fmt]
|
|
s, b = call("POST", "/api/parse", {"file_id": fid})
|
|
d = b.get("data") or {}
|
|
check(f"parse {fmt}", s == 200 and b.get("code") == 0
|
|
and marker in d.get("text", "") and d.get("chars", 0) >= 20,
|
|
str(b)[:80])
|
|
|
|
# ---- URL 抓取分支 ----
|
|
s, b = call("POST", "/api/parse", {"url": "notaurl"})
|
|
check("parse bad url -> 400", s == 400, str(b)[:60])
|
|
s, b = call("POST", "/api/parse", {"url": "http://127.0.0.1:5199/page"})
|
|
d = b.get("data") or {}
|
|
check("parse local url", s == 200 and "UPLOAD_E2E_MARKER_URL" in d.get("text", ""),
|
|
str(b)[:80])
|
|
finally:
|
|
proc.terminate()
|
|
srv.shutdown()
|
|
|
|
import shutil
|
|
shutil.rmtree(TMP, ignore_errors=True)
|
|
print("\n" + ("UPLOAD E2E ALL PASSED" if not failures
|
|
else "FAILURES: " + "; ".join(failures)))
|
|
sys.exit(1 if failures else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|