"""上传解析 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 = ("" "

UPLOAD_E2E_MARKER_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"t" b"

UPLOAD_E2E_MARKER_URL

web capture ok

") 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()