Quentin
9b0910ccb1
后端 (book_mcp_server.py, 335 行, 15 个 MCP 工具):
- create_book / add_document / mark_html_ready / mark_json_ready / mark_failed
- update_book_status / soft_delete_book / retry_failed
- get_book / list_books / get_document / list_documents / get_document_url
- upload_to_rustfs / health
- 走 JSON-RPC + stateless_http + json_response（前端可直接调）
- launchd 守护 / 6 秒拉起（macOS）
- systemd 守护脚本（Linux/codelab）
存储:
- MongoDB (mongo:8.0 容器) - book/book_document/document_file 三表
- RustFS 对象存储（fs.code-lab.cn/scan/）- book/<id>/docs/<doc_id>/{docx,html,json}
- 幂等设计: zip_sha256 唯一 + (book_id+original_path) 唯一
前端 (Vue 3 + TS + Arco Design):
- 列表页 /book/list - 搜索 + 状态过滤 + 进度
- 详情页 /book/:id - 文档折叠 + 文件下载
- MCP 客户端 src/api/mcp.ts (JSON-RPC over fetch)
部署:
- deploy/install.sh (codelab 一键装 backend + systemd)
- deploy/install-frontend.sh (vite dev / 生产构建)
- docs/deploy.md (运维手册)
测试:
- tests/test_basic.py (sanity 7 项)
- tests/test_e2e.py (完整 e2e: docx→HTML→入库→上传→校验)
228 lines
8.9 KiB
Python
228 lines
8.9 KiB
Python
"""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()) |