"""e2e 集成测试 — 完整 Agent 工作流 跑一个真实教辅（1 个 docx → HTML → 入库 → 上传对象存储）。 前提： - docx4j 服务跑在 127.0.0.1:8089 - MongoDB 跑在 127.0.0.1:27018 (book_mcp) - MCP server 跑在 127.0.0.1:8100 """ import asyncio import os import shutil import sys import tempfile import time # 让 book-mcp 包能 import sys.path.insert(0, '/Users/quentin/WorkSpace/book-mcp') from fastmcp.client import Client from services import rustfs_service, wordparse SRC_DOCX = '/tmp/wp2/2027版.曲一线3•2高考淬炼.物理.广西版.一轮复习-突破训练/016-课时作业14 牛顿运动定律（教用）.docx' # 准备 zip（伪造一个只含此 docx 的小 zip） TMP_DIR = '/tmp/book-mcp-e2e' os.makedirs(TMP_DIR, exist_ok=True) TMP_DOCX = os.path.join(TMP_DIR, '016-课时作业14 牛顿运动定律.docx') TMP_HTML = os.path.join(TMP_DIR, '016-课时作业14 牛顿运动定律.html') TMP_JSON = os.path.join(TMP_DIR, '016-课时作业14 牛顿运动定律.json') TMP_ZIP = os.path.join(TMP_DIR, 'sample.zip') # fake zip import zipfile with zipfile.ZipFile(TMP_ZIP, 'w', zipfile.ZIP_DEFLATED) as z: z.write(SRC_DOCX, '课时梯级训练/016-课时作业14 牛顿运动定律.docx') # sha256 import hashlib with open(TMP_ZIP, 'rb') as f: ZIP_SHA = hashlib.sha256(f.read()).hexdigest() def step(n, msg): print(f"\n[{n}] {msg}") def banner(): print("=" * 60) print(f"📚 book-mcp e2e 集成测试") print(f" src docx: {os.path.basename(SRC_DOCX)}") print(f" zip: {TMP_ZIP}") print(f" zip_sha256: {ZIP_SHA[:16]}...") print("=" * 60) async def main(): banner() # ============================================================ # 阶段 1：本地 docx → HTML（用 wordparse 服务） # ============================================================ step(1, '本地 docx → HTML（含自动清洗 + MathJax）') shutil.copy(SRC_DOCX, TMP_DOCX) r = wordparse.convert_one(TMP_DOCX, auto_clean=True) assert r['ok'], f"FAIL: {r}" html = r['html'] with open(TMP_HTML, 'w', encoding='utf-8') as f: f.write(html) print(f" ✓ HTML 生成：{len(html)} chars, " f"公式={r['stats']['formulas']}, " f"图={r['stats']['images']}") # ============================================================ # 阶段 2：本地生成 fake structure.json（Agent 的工作） # ============================================================ step(2, '生成假结构化 JSON（模拟 Agent 解析 HTML）') import json structure = { 'doc_id': 'TBD_BY_MCP', 'original_filename': '016-课时作业14 牛顿运动定律.docx', 'schema_version': '1.0', 'sections': [ {'type': 'title', 'text': '016 课时作业14 牛顿运动定律'}, {'type': 'question', 'no': 1, 'stem': '牛顿第一定律...', 'choices': ['A', 'B', 'C', 'D']}, ], 'stats': r['stats'], } with open(TMP_JSON, 'w', encoding='utf-8') as f: json.dump(structure, f, ensure_ascii=False, indent=2) print(f" ✓ JSON 生成：{os.path.getsize(TMP_JSON)} bytes") # ============================================================ # 阶段 3：MCP 工作流 # ============================================================ step(3, 'MCP 工作流：create_book + add_doc + upload + mark_ready') async with Client('http://127.0.0.1:8100/mcp') as c: # 3.1 create_book unique_suffix = f"{int(time.time())}-{os.getpid()}" r = await c.call_tool('create_book', { 'name': f'2027版.曲一线3•2物理.广西版.一轮复习-突破训练-{unique_suffix}', 'zip_path': f'book/e2e/sample-{unique_suffix}.zip', 'zip_sha256': ZIP_SHA, 'zip_size': os.path.getsize(TMP_ZIP), 'subject': '物理', 'grade': '高三', 'version': f'2027版-{unique_suffix}', 'type': '突破训练', 'created_by': 'e2e-test', }) d = r.data assert d['ok'], f"create_book FAIL: {d}" book_id = d['book_id'] print(f" ✓ create_book id={book_id} ({'幂等' if d['idempotent'] else '新建'})") # 3.2 add_document r = await c.call_tool('add_document', { 'book_id': book_id, 'original_filename': '016-课时作业14 牛顿运动定律.docx', 'original_path': '课时梯级训练/016-课时作业14 牛顿运动定律.docx', }) d = r.data assert d['ok'], f"add_document FAIL: {d}" doc_id = d['doc_id'] print(f" ✓ add_document doc_id={doc_id} type={d['doc_type']}") # 3.3 upload zip（先 create_book.zip） zip_key = f'book/{book_id}/source.zip' zip_upload = rustfs_service.upload(TMP_ZIP, zip_key) print(f" ✓ zip 上传 → {zip_upload['url']}") # 3.4 upload html html_key = rustfs_service.build_key(book_id, doc_id, 'html') html_upload = rustfs_service.upload(TMP_HTML, html_key) print(f" ✓ html 上传 → {html_upload['url']}") # 3.5 mark_html_ready r = await c.call_tool('mark_html_ready', { 'doc_id': doc_id, 'html_url': html_upload['url'], 'html_size': html_upload['size'], 'formulas': r['stats']['formulas'] if False else 0, # 用下面 stats 'images': 0, 'chars': len(html), }) d = r.data assert d['ok'], f"mark_html_ready FAIL: {d}" print(f" ✓ mark_html_ready file_id={d['file_id']}") # 3.6 upload json + mark_json_ready json_key = rustfs_service.build_key(book_id, doc_id, 'json') json_upload = rustfs_service.upload(TMP_JSON, json_key) print(f" ✓ json 上传 → {json_upload['url']}") r = await c.call_tool('mark_json_ready', { 'doc_id': doc_id, 'json_url': json_upload['url'], 'json_size': json_upload['size'], 'schema_version': '1.0', }) d = r.data assert d['ok'] print(f" ✓ mark_json_ready file_id={d['file_id']}") # 3.7 update_book_status r = await c.call_tool('update_book_status', { 'book_id': book_id, 'status': 'done', }) assert d['ok'] print(f" ✓ update_book_status → done") # ============================================================ # 阶段 4：校验数据 # ============================================================ step(4, '校验 get_book 返回完整数据') r = await c.call_tool('get_book', {'book_id': book_id}) b = r.data assert b['name'].startswith('2027版.曲一线') assert b['status'] == 'done' assert b['total_count'] == 1 assert b['success_count'] == 1 assert len(b['documents']) == 1 doc = b['documents'][0] assert doc['doc_id'] == doc_id assert doc['status'] == 'json_ready' assert len(doc['files']) == 2 ft = {f['file_type']: f for f in doc['files']} assert 'html' in ft and 'json' in ft assert ft['html']['object_key'] == html_key assert ft['json']['object_key'] == json_key print(f" ✓ 教辅：name={b['name'][:30]}... status={b['status']}") print(f" ✓ 文档：{doc['original_filename'][:30]}... status={doc['status']}") for f in doc['files']: print(f" {f['file_type']}: {f['url']} ({f['size']} B)") # ============================================================ # 阶段 5：校验对象存储能下载 # ============================================================ step(5, '校验对象存储可访问') import urllib.request for ftype in ['html', 'json']: url = ft[ftype]['url'] try: with urllib.request.urlopen(url, timeout=10) as resp: body = resp.read() print(f" ✓ {ftype} GET {url[:60]}... → {len(body)} B (HTTP {resp.status})") assert len(body) > 100 except Exception as e: print(f" ✗ {ftype} GET FAIL: {e}") raise # ============================================================ # 阶段 6：list_books / health # ============================================================ step(6, 'list_books + health') r = await c.call_tool('list_books', {'page': 1, 'page_size': 3, 'keyword': '物理'}) print(f" ✓ list_books(物理): total={r.data['total']}") for bk in r.data['books']: print(f" id={bk['id']} {bk['name'][:40]} [{bk['status']}]") r = await c.call_tool('health', {}) print(f" ✓ health: {r.data}") print("\n" + "=" * 60) print("✅ e2e 集成测试全流程通过") print("=" * 60) if __name__ == '__main__': asyncio.run(main())