699 lines
27 KiB
Python
699 lines
27 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
OCS × AI 中间件 · 图片题视觉链路 E2E 测试脚手架（独立测试，不改产品代码）
|
||
|
||
背景
|
||
----
|
||
OCS 刷课脚本会把题目/选项的图片 URL 以纯文本内嵌进 title/options，例如：
|
||
title = '<img src="https://p.ananas.chaoxing.com/star3/origin/xxx.png" />'
|
||
title = '求x：https://p.ananas.chaoxing.com/star3/origin/xxx.png'
|
||
options = 'https://.../a.png\nhttps://.../b.png\n...' （选项整行都是图片）
|
||
中间件的目标链路：识别这些 URL → 带 chaoxing Referer 下载图片 → 转 base64（data URL）
|
||
→ 交给 DeepSeek 视觉模型 → 返回选项字母。
|
||
|
||
本脚本提供：
|
||
1) 纯离线单测：直接调用 ai_client 里的图片 URL 抽取函数，验证 HTML img / markdown /
|
||
带扩展名裸 URL / chaoxing 无扩展名 URL 都能抽出，且去重、顺序稳定。不联网、不花钱。
|
||
2) 真实 E2E：PIL 生成英文数字测试图，起本地 ThreadingHTTPServer（随机端口，绝不占用 8777），
|
||
通过 fastapi.testclient.TestClient 走真实 POST /api/search，验证 3 个图片题场景：
|
||
A) title 内嵌 HTML <img>（该图片服务端要求 Referer 含 chaoxing，否则 403，用于验证防盗链下载）
|
||
B) title 是裸图片 URL 文本
|
||
C) options 四行都是图片 URL（图片内容分别是 3/5/7/9）
|
||
每个场景期望 answer 归一化后为 "B"（3x + 5 = 20 → x = 5 → 选项 B）。
|
||
注意：这会真实调用 DeepSeek API（金额很小，已获允许）。
|
||
|
||
用法
|
||
----
|
||
# 在项目根目录执行（也可在其他目录执行，脚本会自动切回项目根目录）
|
||
$env:PYTHONIOENCODING='utf-8'
|
||
py tests/test_vision_e2e.py # 完整：离线单测 + 3 个真实接口场景
|
||
py tests/test_vision_e2e.py --offline-only # 只跑离线单测（不联网、不花钱、不开图片服务）
|
||
py tests/test_vision_e2e.py --e2e-only # 只跑真实接口 E2E
|
||
py tests/test_vision_e2e.py --save-fixtures # 额外把生成的测试图存到 tests/fixtures/ 便于人工检查
|
||
py tests/test_vision_e2e.py --baseline-text-only
|
||
# 基线诊断：运行时临时置空 vision_model（模拟修复前的纯文字链路，
|
||
# 不改 config.yaml），用于证明本测试确实能捕获「图片题答不出」的故障。
|
||
# 该模式预期 FAIL、退出码 1。
|
||
|
||
退出码：全部通过 0，否则 1（便于 CI/流水线判断）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import importlib
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
import traceback
|
||
from datetime import datetime
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from urllib.parse import urlsplit
|
||
|
||
# ────────────────────────────── 路径 ──────────────────────────────
|
||
# tests/ 的上一级即项目根目录（server.py / ai_client.py / config.yaml 所在处）。
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
FIXTURES_DIR = Path(__file__).resolve().parent / "fixtures"
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
# ────────────────────────────── 期望答案工具 ──────────────────────────────
|
||
|
||
_LETTER_RUN_RE = re.compile(r"(?<![A-Z])([A-H])(?![A-Z])")
|
||
|
||
|
||
def normalize_answer(raw) -> str:
|
||
"""把模型返回的 answer 归一化成选项字母。
|
||
|
||
规则（尽力而为，失败时返回空串）：
|
||
1. 优先抓取「独立的」大写选项字母（A–H），如 'B'、'B. 5'、'答案是 B'、'The answer is B'
|
||
→ 返回 'B'；多选返回 'B#D'。
|
||
2. 否则退化为「去空白/标点、只保留大写字母」，如 ' x = 5 ' → ''（不是选项字母，判失败）。
|
||
"""
|
||
if raw is None:
|
||
return ""
|
||
text = str(raw).strip().upper()
|
||
if not text:
|
||
return ""
|
||
letters = _LETTER_RUN_RE.findall(text)
|
||
if letters:
|
||
uniq = list(dict.fromkeys(letters))
|
||
return "#".join(uniq) if len(uniq) > 1 else uniq[0]
|
||
return re.sub(r"[^A-Z]", "", text)
|
||
|
||
|
||
# ────────────────────────────── 测试图片生成（PIL） ──────────────────────────────
|
||
#
|
||
# 只用英文/数字，避免 Windows 缺少中文字体。
|
||
# 图片内容刻意不放中文，字体优先用 Windows 自带的 Arial/Segoe UI。
|
||
FONT_CANDIDATES = [
|
||
r"C:\Windows\Fonts\arialbd.ttf",
|
||
r"C:\Windows\Fonts\arial.ttf",
|
||
r"C:\Windows\Fonts\segoeuib.ttf",
|
||
r"C:\Windows\Fonts\segoeui.ttf",
|
||
r"C:\Windows\Fonts\calibrib.ttf",
|
||
r"C:\Windows\Fonts\calibri.ttf",
|
||
r"C:\Windows\Fonts\DejaVuSans.ttf",
|
||
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
|
||
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
||
]
|
||
|
||
# 题目图片内容；答案为 x = 5，对应选项 B。
|
||
QUESTION_TEXT = "Solve for x: 3x + 5 = 20"
|
||
OPTION_TEXTS = "A. 3\nB. 5\nC. 7\nD. 9"
|
||
EXPECTED_LETTER = "B"
|
||
|
||
|
||
def _load_font(size: int):
|
||
"""按候选列表加载 TrueType 字体，失败时退回 Pillow 内置字体。"""
|
||
from PIL import ImageFont
|
||
|
||
for path in FONT_CANDIDATES:
|
||
if os.path.isfile(path):
|
||
try:
|
||
return ImageFont.truetype(path, size)
|
||
except Exception:
|
||
continue
|
||
try: # Pillow >= 10.1 支持给内置字体指定字号
|
||
return ImageFont.load_default(size=size)
|
||
except TypeError:
|
||
return ImageFont.load_default()
|
||
|
||
|
||
def render_text_png(
|
||
lines,
|
||
font_size: int = 52,
|
||
line_gap: int = 18,
|
||
margin: int = 40,
|
||
) -> bytes:
|
||
"""把若干行文字渲染成白底黑字 PNG（返回 bytes）。"""
|
||
from PIL import Image, ImageDraw
|
||
|
||
font = _load_font(font_size)
|
||
probe = Image.new("RGB", (8, 8), "white")
|
||
probe_draw = ImageDraw.Draw(probe)
|
||
|
||
line_sizes = []
|
||
widths, heights = [], []
|
||
for line in lines:
|
||
bbox = probe_draw.textbbox((0, 0), line, font=font)
|
||
w, h = bbox[2] - bbox[0], bbox[3] - bbox[1]
|
||
widths.append(w)
|
||
heights.append(h)
|
||
line_sizes.append((w, h))
|
||
|
||
width = max(widths) + margin * 2
|
||
height = sum(heights) + line_gap * max(0, len(lines) - 1) + margin * 2
|
||
width = max(width, 16)
|
||
height = max(height, 16)
|
||
|
||
img = Image.new("RGB", (width, height), "white")
|
||
draw = ImageDraw.Draw(img)
|
||
y = margin
|
||
for line, (w, h) in zip(lines, line_sizes):
|
||
# 水平居中，视觉上更像正常题图
|
||
draw.text(((width - w) // 2, y), line, font=font, fill="black")
|
||
y += h + line_gap
|
||
|
||
buf = io.BytesIO()
|
||
img.save(buf, format="PNG", optimize=True)
|
||
return buf.getvalue()
|
||
|
||
|
||
def build_test_images() -> dict:
|
||
"""生成所有测试图片：路径 → PNG bytes。"""
|
||
question_png = render_text_png([QUESTION_TEXT], font_size=56, margin=48)
|
||
option_pngs = {
|
||
digit: render_text_png([digit], font_size=84, margin=56) for digit in ("3", "5", "7", "9")
|
||
}
|
||
return {
|
||
# 场景 A：防盗链图片（服务端校验 Referer 含 chaoxing）
|
||
"/strict/q_a.png": question_png,
|
||
# 场景 B：同一道题的普通图片（同内容、不同 URL，用来隔离「抽取裸 URL」能力）
|
||
"/plain/q_b.png": question_png,
|
||
# 场景 C：四个选项图片
|
||
"/plain/opt_3.png": option_pngs["3"],
|
||
"/plain/opt_5.png": option_pngs["5"],
|
||
"/plain/opt_7.png": option_pngs["7"],
|
||
"/plain/opt_9.png": option_pngs["9"],
|
||
}
|
||
|
||
|
||
STRICT_PATHS = {"/strict/q_a.png"}
|
||
|
||
|
||
# ────────────────────────────── 本地图片 HTTP 服务 ──────────────────────────────
|
||
|
||
|
||
class _RequestLog:
|
||
"""记录图片服务收到的请求，用于诊断 Referer / 下载行为。"""
|
||
|
||
def __init__(self):
|
||
self._lock = threading.Lock()
|
||
self.entries = []
|
||
|
||
def add(self, entry: dict) -> None:
|
||
with self._lock:
|
||
self.entries.append(entry)
|
||
|
||
def snapshot(self) -> list:
|
||
with self._lock:
|
||
return list(self.entries)
|
||
|
||
|
||
def start_image_server(images: dict, strict_paths: set):
|
||
"""在 127.0.0.1 的随机端口启动图片服务。
|
||
|
||
- strict_paths 中的路径：请求头 Referer 必须含 "chaoxing"（不区分大小写），否则返回 403。
|
||
- 其余路径直接返回对应 PNG。
|
||
返回 (server, log, port)；调用方负责 server.shutdown() / server.server_close()。
|
||
"""
|
||
log = _RequestLog()
|
||
|
||
class ImageHandler(BaseHTTPRequestHandler):
|
||
server_version = "OCSTestImageServer/1.0"
|
||
sys_version = ""
|
||
|
||
def log_message(self, fmt, *args): # 静默，避免刷屏
|
||
return
|
||
|
||
def _send(self, status: int, body: bytes = b"", content_type: str = "text/plain; charset=utf-8", head_only: bool = False):
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
if not head_only and body:
|
||
try:
|
||
self.wfile.write(body)
|
||
except (BrokenPipeError, ConnectionResetError):
|
||
pass
|
||
|
||
def _handle(self, head_only: bool = False):
|
||
path = urlsplit(self.path).path
|
||
referer = self.headers.get("Referer") or ""
|
||
user_agent = self.headers.get("User-Agent") or ""
|
||
|
||
status = 404
|
||
body = b"not found"
|
||
content_type = "text/plain; charset=utf-8"
|
||
|
||
if path in images:
|
||
if path in strict_paths and "chaoxing" not in referer.lower():
|
||
# 模拟超星防盗链：缺少 chaoxing Referer 一律 403
|
||
status = 403
|
||
body = b"403 Forbidden: image requires Referer containing 'chaoxing'"
|
||
else:
|
||
status = 200
|
||
body = images[path]
|
||
content_type = "image/png"
|
||
elif path == "/_stats":
|
||
status = 200
|
||
body = json.dumps({"request_count": len(log.snapshot())}).encode("utf-8")
|
||
content_type = "application/json; charset=utf-8"
|
||
|
||
log.add(
|
||
{
|
||
"path": path,
|
||
"referer": referer,
|
||
"user_agent": user_agent[:120],
|
||
"status": status,
|
||
"time": time.strftime("%H:%M:%S"),
|
||
}
|
||
)
|
||
self._send(status, body, content_type, head_only)
|
||
|
||
def do_GET(self): # noqa: N802
|
||
self._handle()
|
||
|
||
def do_HEAD(self): # noqa: N802
|
||
self._handle(head_only=True)
|
||
|
||
server = ThreadingHTTPServer(("127.0.0.1", 0), ImageHandler)
|
||
server.daemon_threads = True
|
||
thread = threading.Thread(target=server.serve_forever, name="test-image-server", daemon=True)
|
||
thread.start()
|
||
port = server.server_address[1]
|
||
return server, log, port
|
||
|
||
|
||
# ────────────────────────────── 离线单测：图片 URL 抽取 ──────────────────────────────
|
||
|
||
# 依次尝试这些函数名；若 Lead 的最终实现换了名字，把新名字加到最前面即可。
|
||
EXTRACTOR_NAMES = (
|
||
"_extract_image_urls",
|
||
"extract_image_urls",
|
||
"_extract_img_urls",
|
||
"extract_img_urls",
|
||
"_extract_image_urls_from_text",
|
||
"extract_image_urls_from_text",
|
||
"_find_image_urls",
|
||
"find_image_urls",
|
||
"_collect_image_urls",
|
||
"collect_image_urls",
|
||
)
|
||
|
||
|
||
def resolve_extractor():
|
||
"""定位 ai_client 中对外暴露的「图片 URL 抽取」函数。
|
||
|
||
返回 ((module_name, func_name, func), 描述) ；找不到时返回 (None, 失败原因)。
|
||
"""
|
||
notes = []
|
||
|
||
try:
|
||
import ai_client as mod
|
||
except Exception as exc: # pragma: no cover - 环境异常
|
||
return None, f"导入 ai_client 失败: {exc!r}"
|
||
|
||
for name in EXTRACTOR_NAMES:
|
||
fn = getattr(mod, name, None)
|
||
if callable(fn):
|
||
return (mod.__name__, name, fn), f"{mod.__name__}.{name}"
|
||
|
||
notes.append("ai_client 中未找到候选函数: " + " / ".join(EXTRACTOR_NAMES))
|
||
|
||
# 兜底：扫描项目根目录其他 .py（比如 Lead 抽出了 image_utils.py），找同名 def。
|
||
for py in sorted(PROJECT_ROOT.glob("*.py")):
|
||
if py.name == Path(__file__).name or py.name.startswith("_probe") or py.name.startswith("_test"):
|
||
continue
|
||
try:
|
||
source = py.read_text(encoding="utf-8")
|
||
except Exception:
|
||
continue
|
||
for name in EXTRACTOR_NAMES:
|
||
if re.search(rf"^def\s+{re.escape(name)}\s*\(", source, re.M):
|
||
try:
|
||
alt_mod = importlib.import_module(py.stem)
|
||
except Exception as exc:
|
||
notes.append(f"{py.name}: 导入失败 {exc!r}")
|
||
continue
|
||
fn = getattr(alt_mod, name, None)
|
||
if callable(fn):
|
||
return (alt_mod.__name__, name, fn), f"{alt_mod.__name__}.{name}（来自 {py.name}）"
|
||
|
||
return None, "；".join(notes)
|
||
|
||
|
||
# (用例名, 输入文本, 期望抽出的 URL 列表[含顺序])
|
||
OFFLINE_CASES = [
|
||
(
|
||
"HTML <img src=...>",
|
||
'<p>题目：<img src="https://p.ananas.chaoxing.com/star3/origin/html_1.png" /></p>',
|
||
["https://p.ananas.chaoxing.com/star3/origin/html_1.png"],
|
||
),
|
||
(
|
||
"Markdown ",
|
||
"题图 ",
|
||
["https://example.com/imgs/md_1.jpg"],
|
||
),
|
||
(
|
||
"裸 URL（带扩展名）",
|
||
"求x：https://example.com/imgs/bare_1.png 谢谢",
|
||
["https://example.com/imgs/bare_1.png"],
|
||
),
|
||
(
|
||
"chaoxing 无扩展名 URL",
|
||
"请查看题图：https://p.ananas.chaoxing.com/star3/origin/abc123 这是题干",
|
||
["https://p.ananas.chaoxing.com/star3/origin/abc123"],
|
||
),
|
||
(
|
||
"去重 + 顺序稳定",
|
||
"A https://e.com/i/1.png B https://e.com/i/2.jpg C https://e.com/i/1.png",
|
||
["https://e.com/i/1.png", "https://e.com/i/2.jpg"],
|
||
),
|
||
]
|
||
|
||
|
||
def run_offline_tests() -> tuple:
|
||
"""跑离线抽取单测，返回 (passed, total)。"""
|
||
print("\n[离线] 图片 URL 抽取单测（不联网、不花钱）")
|
||
resolved, description = resolve_extractor()
|
||
if resolved is None:
|
||
print(f" [FAIL] 找不到图片 URL 抽取函数。{description}")
|
||
print(f" 候选函数名: {', '.join(EXTRACTOR_NAMES)}")
|
||
print(" 若最终实现改名/换模块，请更新 EXTRACTOR_NAMES 后复跑。")
|
||
return 0, len(OFFLINE_CASES)
|
||
|
||
_module_name, _func_name, fn = resolved
|
||
print(f" 使用函数: {description}")
|
||
|
||
passed = 0
|
||
for label, text, expected in OFFLINE_CASES:
|
||
try:
|
||
got = fn(text)
|
||
if got is None:
|
||
got = []
|
||
got = [str(item) for item in list(got)]
|
||
except Exception as exc:
|
||
print(f" [FAIL] {label} — 调用异常: {exc!r}")
|
||
traceback.print_exc(limit=2)
|
||
continue
|
||
|
||
if got == expected:
|
||
passed += 1
|
||
print(f" [PASS] {label}")
|
||
else:
|
||
print(f" [FAIL] {label}")
|
||
print(f" 输入: {text}")
|
||
print(f" 期望: {expected}")
|
||
print(f" 实际: {got}")
|
||
|
||
print(f" 离线单测: PASS {passed}/{len(OFFLINE_CASES)}")
|
||
return passed, len(OFFLINE_CASES)
|
||
|
||
|
||
# ────────────────────────────── E2E：真实接口 ──────────────────────────────
|
||
|
||
|
||
def build_scenarios(port: int) -> list:
|
||
"""构造 3 个场景的 /api/search 请求体。"""
|
||
base = f"http://127.0.0.1:{port}"
|
||
single = "single"
|
||
return [
|
||
{
|
||
"id": "A",
|
||
"label": "title 内嵌 HTML <img>（图片服务端校验 chaoxing Referer）",
|
||
"payload": {
|
||
"title": f'<img src="{base}/strict/q_a.png" />',
|
||
"type": single,
|
||
"options": OPTION_TEXTS,
|
||
},
|
||
"expected": EXPECTED_LETTER,
|
||
},
|
||
{
|
||
"id": "B",
|
||
"label": "title 为裸图片 URL 文本",
|
||
"payload": {
|
||
"title": f"{base}/plain/q_b.png",
|
||
"type": single,
|
||
"options": OPTION_TEXTS,
|
||
},
|
||
"expected": EXPECTED_LETTER,
|
||
},
|
||
{
|
||
"id": "C",
|
||
"label": "options 四行均为图片 URL（图片内容 3/5/7/9）",
|
||
"payload": {
|
||
"title": QUESTION_TEXT,
|
||
"type": single,
|
||
"options": "\n".join(
|
||
f"{base}/plain/opt_{digit}.png" for digit in ("3", "5", "7", "9")
|
||
),
|
||
},
|
||
"expected": EXPECTED_LETTER,
|
||
},
|
||
]
|
||
|
||
|
||
def run_e2e(scenarios: list) -> list:
|
||
"""通过 TestClient 真实调用 /api/search，返回每个场景的结果列表。"""
|
||
print("\n[E2E] 通过 FastAPI TestClient 真实调用 POST /api/search（会真实消耗少量 DeepSeek 额度）")
|
||
|
||
try:
|
||
from fastapi.testclient import TestClient
|
||
from server import app
|
||
except Exception as exc:
|
||
print(f" [FATAL] 导入 server.app / TestClient 失败: {exc!r}")
|
||
traceback.print_exc(limit=3)
|
||
return [
|
||
{"id": sc["id"], "ok": False, "code": None, "raw": "", "norm": "", "expected": sc["expected"], "error": repr(exc)}
|
||
for sc in scenarios
|
||
]
|
||
|
||
results = []
|
||
# 必须用 with，才会触发 FastAPI lifespan（里面才创建 ai_client）
|
||
with TestClient(app) as client:
|
||
for sc in scenarios:
|
||
payload = sc["payload"]
|
||
print(f"\n [场景 {sc['id']}] {sc['label']}")
|
||
print(f" title : {payload['title']!r}")
|
||
print(f" options: {payload['options']!r}")
|
||
print(f" type : {payload['type']}")
|
||
|
||
t0 = time.time()
|
||
data = {}
|
||
error = None
|
||
http_status = None
|
||
try:
|
||
resp = client.post("/api/search", json=payload)
|
||
elapsed = time.time() - t0
|
||
http_status = resp.status_code
|
||
try:
|
||
data = resp.json()
|
||
except Exception:
|
||
data = {"_raw_text": resp.text[:800]}
|
||
print(f" POST /api/search → HTTP {http_status}，耗时 {elapsed:.1f}s")
|
||
except Exception as exc:
|
||
elapsed = time.time() - t0
|
||
error = repr(exc)
|
||
print(f" POST /api/search 抛异常（耗时 {elapsed:.1f}s）: {error}")
|
||
traceback.print_exc(limit=3)
|
||
|
||
code = data.get("code") if isinstance(data, dict) else None
|
||
raw = data.get("answer", "") if isinstance(data, dict) else ""
|
||
msg = data.get("msg", "") if isinstance(data, dict) else ""
|
||
norm = normalize_answer(raw)
|
||
|
||
if code == 1:
|
||
print(f" code=1 answer={raw!r}")
|
||
else:
|
||
print(f" code={code!r} msg={msg!r} 原始响应={str(data)[:500]}")
|
||
|
||
ok = bool(http_status == 200 and code == 1 and norm == sc["expected"])
|
||
print(f" 归一化 answer={norm!r}，期望={sc['expected']!r} → {'PASS' if ok else 'FAIL'}")
|
||
|
||
results.append(
|
||
{
|
||
"id": sc["id"],
|
||
"ok": ok,
|
||
"code": code,
|
||
"raw": raw,
|
||
"norm": norm,
|
||
"expected": sc["expected"],
|
||
"http_status": http_status,
|
||
"error": error,
|
||
}
|
||
)
|
||
return results
|
||
|
||
|
||
def run_referer_check(log: _RequestLog, strict_paths: set) -> bool:
|
||
"""校验中间件下载防盗链图片时确实带了 chaoxing Referer。"""
|
||
print("\n[Referer] 防盗链校验（仅 Referer 含 chaoxing 的请求才能 200）")
|
||
entries = log.snapshot()
|
||
if not entries:
|
||
print(" [FAIL] 图片服务没有收到任何请求 —— 说明中间件没有下载图片（视觉/OCR 链路未生效）")
|
||
return False
|
||
|
||
print(" 图片服务请求记录:")
|
||
for entry in entries:
|
||
marker = "✓" if entry["status"] == 200 else "✗"
|
||
print(
|
||
f" {marker} GET {entry['path']} → {entry['status']} | "
|
||
f"Referer={entry['referer']!r} | UA={entry['user_agent'][:60]!r}"
|
||
)
|
||
|
||
strict_entries = [e for e in entries if e["path"] in strict_paths]
|
||
if not strict_entries:
|
||
print(f" [FAIL] 未请求防盗链图片 {sorted(strict_paths)} —— 无法验证 Referer")
|
||
return False
|
||
|
||
good = [
|
||
e
|
||
for e in strict_entries
|
||
if e["status"] == 200 and "chaoxing" in (e["referer"] or "").lower()
|
||
]
|
||
if good:
|
||
print(f" [PASS] 防盗链图片被成功下载 {len(good)} 次（下载时带了 chaoxing Referer）")
|
||
return True
|
||
|
||
print(" [FAIL] 防盗链图片全部被 403 拒绝 —— 中间件下载图片时没有带 chaoxing Referer")
|
||
return False
|
||
|
||
|
||
# ────────────────────────────── main ──────────────────────────────
|
||
|
||
|
||
def _print_header(mode: str) -> None:
|
||
try:
|
||
import PIL
|
||
|
||
pillow_version = PIL.__version__
|
||
except Exception:
|
||
pillow_version = "?"
|
||
|
||
print("=" * 72)
|
||
print("OCS × AI 中间件 · 图片题视觉链路 E2E 测试")
|
||
print(f"项目根目录: {PROJECT_ROOT}")
|
||
print(f"Python: {sys.version.split()[0]} | Pillow: {pillow_version}")
|
||
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"模式: {mode}")
|
||
print("=" * 72)
|
||
|
||
|
||
def _save_fixtures(images: dict) -> None:
|
||
FIXTURES_DIR.mkdir(parents=True, exist_ok=True)
|
||
for path, payload in images.items():
|
||
out = FIXTURES_DIR / Path(path).name
|
||
out.write_bytes(payload)
|
||
print(f"[图片] 测试图片已保存到 {FIXTURES_DIR}")
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
# Windows 控制台中文/符号输出保护
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
parser = argparse.ArgumentParser(description="OCS × AI 中间件图片题 E2E 测试")
|
||
parser.add_argument("--offline-only", action="store_true", help="只跑离线 URL 抽取单测（不联网、不花钱）")
|
||
parser.add_argument("--e2e-only", action="store_true", help="跳过离线单测，只跑真实接口 E2E")
|
||
parser.add_argument("--save-fixtures", action="store_true", help="把生成的测试图片保存到 tests/fixtures/")
|
||
parser.add_argument(
|
||
"--baseline-text-only",
|
||
action="store_true",
|
||
help="基线诊断：运行时临时置空 vision_model（模拟修复前纯文字链路，不改 config.yaml），用于验证本测试能捕获图片题失败",
|
||
)
|
||
args = parser.parse_args(argv)
|
||
|
||
if args.offline_only and args.e2e_only:
|
||
print("[错误] --offline-only 和 --e2e-only 不能同时使用")
|
||
return 2
|
||
if args.baseline_text_only and args.offline_only:
|
||
print("[错误] --baseline-text-only 需要跑 E2E，不能和 --offline-only 同时使用")
|
||
return 2
|
||
|
||
# server.load_config("config.yaml") 使用相对路径，必须保证 CWD 在项目根目录。
|
||
os.chdir(PROJECT_ROOT)
|
||
|
||
if args.offline_only:
|
||
mode = "仅离线单测（不联网）"
|
||
elif args.baseline_text_only:
|
||
mode = "失败基线诊断（临时关闭视觉模型，模拟修复前纯文字链路；预期 FAIL）"
|
||
elif args.e2e_only:
|
||
mode = "仅真实接口 E2E（会消耗 DeepSeek 额度）"
|
||
else:
|
||
mode = "完整（离线单测 + 真实接口 E2E）"
|
||
_print_header(mode)
|
||
|
||
# 生成测试图片（--save-fixtures 时无论哪种模式都落盘，方便人工检查）
|
||
images = None
|
||
if args.save_fixtures or not args.offline_only:
|
||
images = build_test_images()
|
||
if args.save_fixtures:
|
||
_save_fixtures(images)
|
||
|
||
offline_passed = offline_total = 0
|
||
if not args.e2e_only:
|
||
offline_passed, offline_total = run_offline_tests()
|
||
|
||
e2e_results = []
|
||
e2e_total = 0
|
||
referer_ok = None
|
||
if not args.offline_only:
|
||
if args.baseline_text_only:
|
||
# 仅改运行期内存的 CONFIG（不改文件），模拟修复前「没有视觉模型」的纯文字链路。
|
||
import server as _server
|
||
|
||
_server.CONFIG["ai"]["vision_model"] = ""
|
||
print(
|
||
"\n[基线] 已临时置空 server.CONFIG['ai']['vision_model']"
|
||
"（模拟修复前纯文字链路；config.yaml 未被修改）"
|
||
)
|
||
|
||
server, log, port = start_image_server(images, STRICT_PATHS)
|
||
print(f"\n[启动] 本地图片服务: http://127.0.0.1:{port}")
|
||
print(f" 防盗链图片: {sorted(STRICT_PATHS)}（Referer 含 chaoxing 才返回 200）")
|
||
try:
|
||
scenarios = build_scenarios(port)
|
||
e2e_total = len(scenarios)
|
||
e2e_results = run_e2e(scenarios)
|
||
referer_ok = run_referer_check(log, STRICT_PATHS)
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
print("\n[关闭] 本地图片服务已停止")
|
||
|
||
# ── 汇总 ──
|
||
e2e_passed = sum(1 for r in e2e_results if r.get("ok"))
|
||
overall_ok = True
|
||
|
||
print("\n" + "=" * 72)
|
||
print("汇总")
|
||
if not args.e2e_only:
|
||
offline_ok = offline_passed == offline_total
|
||
overall_ok = overall_ok and offline_ok
|
||
print(f" 离线 URL 抽取: PASS {offline_passed}/{offline_total} → {'PASS' if offline_ok else 'FAIL'}")
|
||
if not args.offline_only:
|
||
e2e_ok = e2e_total > 0 and e2e_passed == e2e_total
|
||
overall_ok = overall_ok and e2e_ok
|
||
print(f" E2E 真实接口: PASS {e2e_passed}/{e2e_total} → {'PASS' if e2e_ok else 'FAIL'}")
|
||
if e2e_total:
|
||
print(" 各场景:", " ".join(f"{r['id']}={'PASS' if r['ok'] else 'FAIL'}" for r in e2e_results))
|
||
if referer_ok is not None:
|
||
overall_ok = overall_ok and referer_ok
|
||
print(f" Referer 防盗链: {'PASS' if referer_ok else 'FAIL'}")
|
||
print(f" 总体: {'PASS ✅' if overall_ok else 'FAIL ❌'}")
|
||
if args.baseline_text_only:
|
||
if not overall_ok:
|
||
print(" [基线结论] 已复现修复前故障：关闭视觉链路后图片题 FAIL → 说明本测试能捕获该问题。")
|
||
else:
|
||
print(" [基线结论] 警告：关闭视觉链路后仍全部 PASS，测试可能没有真正覆盖图片视觉链路！")
|
||
elif not overall_ok:
|
||
print(" 提示：若产品代码尚未接入「下载图片 → base64 → 视觉模型」，以上 FAIL 即预期基线。")
|
||
print("=" * 72)
|
||
|
||
return 0 if overall_ok else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|