769 lines
35 KiB
Python
769 lines
35 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
OCS 5.0「图片题优化」契约测试（纯离线，不联网、不花钱）
|
||
|
||
目标
|
||
----
|
||
独立核对三件事：
|
||
1. generate_ocs_config.generate_config() 生成的题库配置，一定能让 OCS 5.0 的
|
||
isAnswererWrapperSupportImageOptimize() 返回 true（即不再弹「当前题库配置暂不支持
|
||
图片、公式等题目处理」警告）；
|
||
2. OCS 5.0 运行时实际发出的 POST JSON 结构，与 ai_client / server 的实现假设一致
|
||
（images 元素是 data:image/png;base64,...；suggestion_* 用 " [图片N] " 带空格；
|
||
text-only 时 images=[]、suggestion_* 为空字符串）；
|
||
3. server.py 对新字段的解析（缺失/类型异常）与 ai_client 的容错行为符合预期。
|
||
|
||
实现方式
|
||
--------
|
||
* 为避免「用自己的 Python 复刻来验证自己的假设」，本测试用 Node 真实执行
|
||
tests/ocs5_emulator.mjs —— 该文件逐段移植自本地 OCS 5.0 源码（函数来源写在文件头），
|
||
保留 JS 的 RegExp / Function / JSON 等真实语义；
|
||
* 配置检测与 payload 构建结果直接来自 emulator；Python 侧只做断言和对接；
|
||
* server 端用 TestClient + 假 AI 客户端（不调用 DeepSeek），只验证字段解析链路；
|
||
* 最后对 OCS 源码文本做静态断言，确认 emulator 移植的关键逻辑与源码字符串一致
|
||
（源码目录不存在时这些检查标记 SKIP，不影响其他检查）。
|
||
|
||
运行（项目根目录）：
|
||
$env:PYTHONIOENCODING='utf-8'
|
||
py tests/test_ocs5_contract.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import io
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# ────────────────────────── 基础设置 ──────────────────────────
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
TESTS_DIR = Path(__file__).resolve().parent
|
||
EMULATOR = TESTS_DIR / "ocs5_emulator.mjs"
|
||
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT))
|
||
os.chdir(ROOT)
|
||
|
||
# OCS 5.0 源码目录：可用环境变量 OCS5_SRC 覆盖
|
||
OCS_SRC_CANDIDATES = [
|
||
os.environ.get("OCS5_SRC", ""),
|
||
r"C:\Users\SiF4.PETRI-DISH\AppData\Local\Temp\ocs_src\ocsjs_exp_5.0\ocsjs-5.0",
|
||
r"C:\Users\SiF4~1.PET\AppData\Local\Temp\ocs_src\ocsjs_exp_5.0\ocsjs-5.0",
|
||
]
|
||
|
||
FAILS = []
|
||
SKIPS = []
|
||
|
||
|
||
def record(name: str, ok: bool, detail: str = "") -> bool:
|
||
print((" [PASS] " if ok else " [FAIL] ") + name + (f" | {detail}" if detail else ""))
|
||
if not ok:
|
||
FAILS.append(name)
|
||
return ok
|
||
|
||
|
||
def skip(name: str, reason: str) -> None:
|
||
print(f" [SKIP] {name} | {reason}")
|
||
SKIPS.append(name)
|
||
|
||
|
||
def find_ocs_src() -> Path | None:
|
||
for cand in OCS_SRC_CANDIDATES:
|
||
if cand and Path(cand).is_dir():
|
||
return Path(cand)
|
||
return None
|
||
|
||
|
||
# ────────────────────────── 测试图片 ──────────────────────────
|
||
FONT_CANDIDATES = [
|
||
r"C:\Windows\Fonts\arial.ttf",
|
||
r"C:\Windows\Fonts\arialbd.ttf",
|
||
r"C:\Windows\Fonts\segoeui.ttf",
|
||
]
|
||
|
||
|
||
def make_png_data_url(text: str, width: int = 700, height: int = 130, font_size: int = 44) -> str:
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
font = None
|
||
for path in FONT_CANDIDATES:
|
||
if os.path.isfile(path):
|
||
try:
|
||
font = ImageFont.truetype(path, font_size)
|
||
break
|
||
except Exception:
|
||
continue
|
||
if font is None:
|
||
font = ImageFont.load_default()
|
||
|
||
img = Image.new("RGB", (width, height), "white")
|
||
draw = ImageDraw.Draw(img)
|
||
draw.text((30, height // 2 - font_size // 2 - 6), text, fill="black", font=font)
|
||
buf = io.BytesIO()
|
||
img.save(buf, format="PNG")
|
||
return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii")
|
||
|
||
|
||
# ────────────────────────── Node emulator 调用 ──────────────────────────
|
||
def node_available() -> tuple:
|
||
node = shutil.which("node")
|
||
return (node is not None, node or "")
|
||
|
||
|
||
def run_emulator(payload: dict) -> dict:
|
||
node_ok, node = node_available()
|
||
if not node_ok:
|
||
raise RuntimeError("未找到 node，可执行文件不可用")
|
||
proc = subprocess.run(
|
||
[node, str(EMULATOR)],
|
||
input=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
||
capture_output=True,
|
||
timeout=180,
|
||
)
|
||
if proc.returncode != 0:
|
||
raise RuntimeError(
|
||
f"emulator 执行失败 rc={proc.returncode}: "
|
||
f"{proc.stderr.decode('utf-8', 'replace')[:2000]}"
|
||
)
|
||
stderr = proc.stderr.decode("utf-8", "replace").strip()
|
||
if stderr:
|
||
print(f" (emulator stderr) {stderr[:300]}")
|
||
return json.loads(proc.stdout.decode("utf-8"))
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 一、题库配置检测（generate_ocs_config → OCS 检测逻辑）
|
||
# ══════════════════════════════════════════════════════════════════
|
||
OLD_CONFIG = [
|
||
{
|
||
"url": "http://127.0.0.1:8777/api/search",
|
||
"name": "AI 智能搜题",
|
||
"method": "post",
|
||
"contentType": "json",
|
||
"data": {"title": "${title}", "type": "${type}", "options": "${options}"},
|
||
"handler": "return (res)=> res.code === 1 ? [res.question, res.answer] : undefined",
|
||
}
|
||
]
|
||
|
||
|
||
def section_config() -> tuple:
|
||
print("\n[一] 题库配置检测（OCS 5.0 isAnswererWrapperSupportImageOptimize）")
|
||
|
||
from generate_ocs_config import generate_config
|
||
|
||
generated = generate_config()
|
||
|
||
# 1.1 generate_config() 与仓库里的 ocs_config.json 是否一致
|
||
config_file = ROOT / "ocs_config.json"
|
||
if config_file.is_file():
|
||
try:
|
||
file_data = json.loads(config_file.read_text(encoding="utf-8"))
|
||
record(
|
||
"generate_config() 与 ocs_config.json 内容一致",
|
||
file_data == generated,
|
||
"" if file_data == generated else "请重新运行 gen_config.bat 生成最新 ocs_config.json",
|
||
)
|
||
except Exception as exc:
|
||
record("ocs_config.json 可解析", False, repr(exc))
|
||
else:
|
||
skip("ocs_config.json 一致性", "文件不存在")
|
||
|
||
# 1.2 用 Node 执行 OCS 原版检测逻辑
|
||
case_wrappers = [
|
||
{
|
||
"name": "旧配置（无图片字段）",
|
||
"wrapper": OLD_CONFIG[0],
|
||
},
|
||
{
|
||
"name": "GET + 占位符（方法不对）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "get",
|
||
"data": {"images": "${images}"}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "method 大写 POST（源码 toLowerCase 后应通过）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "POST",
|
||
"data": {"suggestion_title": "${suggestion_title}"}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "POST 但 data 无任何图片引用（应不支持）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "post",
|
||
"data": {"title": "${title}", "options": "${options}"}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "仅 data.images handler 引用 env.images（应支持）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "post",
|
||
"data": {"images": {"handler": "return (env)=> env.images || []"}}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "handler 中 env.titleImages 不应误判（负向）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "post",
|
||
"data": {"images": {"handler": "return (env)=> env.titleImages || []"}}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "handler 中 env.imagesList 不应误判（负向）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "post",
|
||
"data": {"images": {"handler": "return (env)=> env.imagesList || []"}}, "handler": "return (r)=> r"},
|
||
},
|
||
{
|
||
"name": "解构写法 const { images } = env（应支持）",
|
||
"wrapper": {"url": "http://x/api", "name": "x", "method": "post",
|
||
"data": {"images": {"handler": "return (env)=> { const { images } = env; return images; }"}},
|
||
"handler": "return (r)=> r"},
|
||
},
|
||
]
|
||
|
||
if not node_available()[0]:
|
||
skip("OCS 检测逻辑执行", "未找到 node（需 Node.js 运行 ocs5_emulator.mjs）")
|
||
return generated
|
||
|
||
try:
|
||
detect = run_emulator({"op": "detect", "wrappers": generated, "cases": case_wrappers})
|
||
except Exception as exc:
|
||
record("Node emulator 检测执行", False, repr(exc))
|
||
return generated
|
||
|
||
# 1.3 新配置 / 旧配置的总体判定
|
||
record(
|
||
"新配置通过 OCS 图片题优化检测（警告消失）",
|
||
detect.get("supportsImageOptimize") is True,
|
||
f"per-wrapper={[w['supportsImageOptimize'] for w in detect.get('perWrapper', [])]}",
|
||
)
|
||
old_detect = run_emulator({"op": "detect", "wrappers": OLD_CONFIG})
|
||
record(
|
||
"旧配置会被判定为不支持（复现「暂不支持图片、公式」警告条件）",
|
||
old_detect.get("supportsImageOptimize") is False,
|
||
)
|
||
|
||
# 1.4 每个 wrapper 的检测 + 解析器校验 + 可疑代码扫描
|
||
per = detect.get("perWrapper", [])
|
||
record("新配置每个 wrapper 均通过图片检测", len(per) == 1 and per[0]["supportsImageOptimize"] is True)
|
||
record(
|
||
"新配置通过 OCS AnswerWrapperParser.fromObject 字段校验",
|
||
all(w["parser"]["ok"] for w in per),
|
||
f"errors={[e for w in per for e in w['parser']['errors']]}",
|
||
)
|
||
all_risks = [risks for w in per for risks in w["risks"]]
|
||
record(
|
||
"新配置不会触发 OCS「可疑代码」风险弹窗",
|
||
len(all_risks) == 0,
|
||
f"findings={all_risks}",
|
||
)
|
||
|
||
# 1.5 边界/负向用例
|
||
case_results = {c["name"]: c for c in detect.get("directCases", [])}
|
||
expected = {
|
||
"旧配置（无图片字段）": False,
|
||
"GET + 占位符（方法不对）": False,
|
||
"method 大写 POST（源码 toLowerCase 后应通过）": True,
|
||
"POST 但 data 无任何图片引用（应不支持）": False,
|
||
"仅 data.images handler 引用 env.images（应支持）": True,
|
||
"handler 中 env.titleImages 不应误判（负向）": False,
|
||
"handler 中 env.imagesList 不应误判（负向）": False,
|
||
"解构写法 const { images } = env（应支持）": True,
|
||
}
|
||
for name, want in expected.items():
|
||
got = (case_results.get(name) or {}).get("result")
|
||
record(f"检测边界：{name}", got is want, f"got={got}")
|
||
|
||
# 1.6 粘贴字符串解析路径（AnswerWrapperParser.from(string) 只做 JSON.parse）
|
||
raw = json.dumps(generated, ensure_ascii=False)
|
||
parse = run_emulator({"op": "parse", "wrappers": generated, "rawString": raw})
|
||
record("配置 JSON 字符串可被 OCS 解析", parse.get("stringParseOk") is True)
|
||
record("配置通过 fromObject 校验", parse.get("fromObject", {}).get("ok") is True)
|
||
|
||
return generated
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 二、OCS 运行时 payload 结构（buildAnswererEnv + POST data 解析）
|
||
# ══════════════════════════════════════════════════════════════════
|
||
def section_payload(generated_config: list) -> list:
|
||
print("\n[二] OCS 5.0 运行时 POST JSON 结构（buildAnswererEnv + data 解析）")
|
||
|
||
question_png = make_png_data_url("Solve for x: 3x + 5 = 20")
|
||
digit_pngs = {d: make_png_data_url(d, 220, 150, 64) for d in ("3", "5", "7", "9")}
|
||
|
||
q_url = "https://p.ananas.chaoxing.com/star3/origin/q.png"
|
||
opt_urls = [f"https://p.ananas.chaoxing.com/star3/origin/{d}.png" for d in ("3", "5", "7", "9")]
|
||
|
||
scenarios = [
|
||
{
|
||
"name": "纯文字题",
|
||
"type": "single",
|
||
"title": "2 + 3 = ?",
|
||
"options": "A. 4\nB. 5\nC. 6\nD. 7",
|
||
"titleImages": [],
|
||
"optionsImages": [],
|
||
"imagePayloads": {},
|
||
},
|
||
{
|
||
"name": "title 图片（DOM 顺序 titleImages）",
|
||
"type": "single",
|
||
"title": "Solve for x: 3x + 5 = 20 " + q_url,
|
||
"options": "A. 3\nB. 5\nC. 7\nD. 9",
|
||
"titleImages": [q_url],
|
||
"optionsImages": [],
|
||
"imagePayloads": {q_url: question_png},
|
||
},
|
||
{
|
||
"name": "纯图片选项",
|
||
"type": "single",
|
||
"title": "Solve for x: 3x + 5 = 20",
|
||
"options": "\n".join(f"{letter}. {url}" for letter, url in zip("ABCD", opt_urls)),
|
||
"titleImages": [],
|
||
"optionsImages": list(opt_urls),
|
||
"imagePayloads": {u: digit_pngs[d] for u, d in zip(opt_urls, ("3", "5", "7", "9"))},
|
||
},
|
||
{
|
||
"name": "混合多图（title 1 张 + options 4 张）",
|
||
"type": "single",
|
||
"title": "求 x 的值： " + q_url,
|
||
"options": "\n".join(f"{letter}. {url}" for letter, url in zip("ABCD", opt_urls)),
|
||
"titleImages": [q_url],
|
||
"optionsImages": list(opt_urls),
|
||
"imagePayloads": {
|
||
q_url: question_png,
|
||
**{u: digit_pngs[d] for u, d in zip(opt_urls, ("3", "5", "7", "9"))},
|
||
},
|
||
},
|
||
]
|
||
|
||
if not node_available()[0]:
|
||
skip("OCS payload 构建", "未找到 node")
|
||
return []
|
||
|
||
try:
|
||
built = run_emulator({"op": "build", "wrappers": generated_config, "scenarios": scenarios})
|
||
except Exception as exc:
|
||
record("Node emulator 构建 payload", False, repr(exc))
|
||
return []
|
||
|
||
by_name = {s["name"]: s for s in built.get("scenarios", [])}
|
||
bodies = []
|
||
for scenario in scenarios:
|
||
name = scenario["name"]
|
||
item = by_name.get(name)
|
||
if not item:
|
||
record(f"payload 构建：{name}", False, "emulator 未返回该场景")
|
||
continue
|
||
requests = item.get("requests") or []
|
||
if not requests:
|
||
record(f"payload 构建：{name}", False, "无 wrapper 请求")
|
||
continue
|
||
body = requests[0]["body"]
|
||
bodies.append((name, scenario, body))
|
||
|
||
ok = True
|
||
detail = ""
|
||
# 所有场景公共断言：字段齐全、method 正确、key 与 server 期望一致
|
||
ok = ok and requests[0]["method"] == "post"
|
||
ok = ok and requests[0]["contentType"] == "json"
|
||
ok = ok and isinstance(body.get("images"), list)
|
||
ok = ok and isinstance(body.get("suggestion_title"), str)
|
||
ok = ok and isinstance(body.get("suggestion_options"), str)
|
||
# images 元素必须是 OCS 归一化后的 data:image/png;base64,...
|
||
imgs = body.get("images") or []
|
||
ok = ok and all(
|
||
isinstance(x, str) and x.startswith("data:image/png;base64,") and len(x) > 100 for x in imgs
|
||
)
|
||
if not ok:
|
||
detail = f"method={requests[0]['method']} keys={list(body.keys())} images={len(imgs)}"
|
||
record(f"payload 结构：{name}", ok, detail)
|
||
|
||
# 逐场景精确断言
|
||
if "纯文字题" in by_name:
|
||
body = by_name["纯文字题"]["requests"][0]["body"]
|
||
record(
|
||
"纯文字题：images=[]、suggestion_* 为空字符串（不是 undefined）",
|
||
body["images"] == [] and body["suggestion_title"] == "" and body["suggestion_options"] == "",
|
||
f"images={body['images']!r} title={body['suggestion_title']!r} opts={body['suggestion_options']!r}",
|
||
)
|
||
record(
|
||
"纯文字题：title/type/options 原样保留",
|
||
body["title"] == "2 + 3 = ?" and body["type"] == "single" and body["options"] == "A. 4\nB. 5\nC. 6\nD. 7",
|
||
)
|
||
|
||
if "title 图片（DOM 顺序 titleImages）" in by_name:
|
||
scenario = by_name["title 图片（DOM 顺序 titleImages）"]
|
||
body = scenario["requests"][0]["body"]
|
||
expected_title = ("Solve for x: 3x + 5 = 20 " + q_url).replace(q_url, " [图片1] ")
|
||
record("title 图片：images 恰好 1 张 base64", len(body["images"]) == 1)
|
||
record(
|
||
"title 图片：suggestion_title 用 ' [图片1] ' 原位替换（左右各一空格）",
|
||
body["suggestion_title"] == expected_title,
|
||
f"got={body['suggestion_title']!r}",
|
||
)
|
||
record(
|
||
"title 图片：无选项图片时 suggestion_options 为空字符串",
|
||
body["suggestion_options"] == "",
|
||
f"got={body['suggestion_options']!r}",
|
||
)
|
||
|
||
if "纯图片选项" in by_name:
|
||
body = by_name["纯图片选项"]["requests"][0]["body"]
|
||
expected_options = "\n".join(
|
||
f"{letter}. [图片{i + 1}] " for i, letter in enumerate("ABCD")
|
||
)
|
||
expected_title = "Solve for x: 3x + 5 = 20"
|
||
record("纯图片选项：images 恰好 4 张 base64", len(body["images"]) == 4)
|
||
record(
|
||
"纯图片选项：suggestion_title 无图片时保持原文",
|
||
body["suggestion_title"] == expected_title,
|
||
f"got={body['suggestion_title']!r}",
|
||
)
|
||
record(
|
||
"纯图片选项：suggestion_options 按 A/B/C/D 顺序填 [图片1..4]（带空格）",
|
||
body["suggestion_options"] == expected_options,
|
||
f"got={body['suggestion_options']!r}",
|
||
)
|
||
|
||
if "混合多图（title 1 张 + options 4 张）" in by_name:
|
||
body = by_name["混合多图（title 1 张 + options 4 张）"]["requests"][0]["body"]
|
||
expected_options = "\n".join(
|
||
f"{letter}. [图片{i + 2}] " for i, letter in enumerate("ABCD")
|
||
)
|
||
record(
|
||
"混合多图：编号先 title 后 options（title=[图片1]，options=[图片2..5]）",
|
||
body["suggestion_title"].endswith(" [图片1] ")
|
||
and body["suggestion_options"] == expected_options,
|
||
f"title={body['suggestion_title']!r} opts={body['suggestion_options']!r}",
|
||
)
|
||
|
||
# 交叉验证：OCS 产生的 [图片N] 标记必须能被中间件的解析器识别
|
||
from ai_client import _iter_marker_segments, _prepare_ocs_images
|
||
|
||
segs = _iter_marker_segments("前 [图片12] 后")
|
||
record(
|
||
"中间件能解析 OCS 生成的 ' [图片N] ' 标记",
|
||
[(k, v) for k, v in segs if k == "image"] == [("image", 12)],
|
||
f"segments={segs}",
|
||
)
|
||
record(
|
||
"中间件 _prepare_ocs_images 接受 OCS 的 data URL 数组",
|
||
len(_prepare_ocs_images([question_png, digit_pngs["5"]])) == 2,
|
||
)
|
||
|
||
return bodies
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 三、server 端字段解析（TestClient + 假 AI，不联网）
|
||
# ══════════════════════════════════════════════════════════════════
|
||
class FakeAIClient:
|
||
"""记录 server 传进来的 QuestionInfo，不调用 DeepSeek"""
|
||
|
||
def __init__(self):
|
||
self.questions = []
|
||
|
||
async def ask(self, question):
|
||
self.questions.append(question)
|
||
return "B"
|
||
|
||
|
||
class LogCapture(logging.Handler):
|
||
def __init__(self):
|
||
super().__init__(level=logging.DEBUG)
|
||
self.messages = []
|
||
|
||
def emit(self, record):
|
||
try:
|
||
self.messages.append(record.getMessage())
|
||
except Exception:
|
||
self.messages.append("<log format error>")
|
||
|
||
def reset(self):
|
||
self.messages = []
|
||
|
||
|
||
def section_server(bodies: list) -> None:
|
||
print("\n[三] server.py 新字段解析（TestClient + 假 AI，不联网）")
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
import server as server_module
|
||
|
||
question_png = make_png_data_url("1 + 1 = ?")
|
||
|
||
try:
|
||
client_ctx = TestClient(server_module.app)
|
||
with client_ctx as client:
|
||
fake = FakeAIClient()
|
||
server_module.ai_client = fake
|
||
|
||
health = client.get("/health")
|
||
record("server 启动 /health 正常", health.status_code == 200 and health.json().get("status") == "ok")
|
||
|
||
# 3.1 OCS emulator 产出的真实结构 → server → QuestionInfo
|
||
for name, scenario, body in bodies:
|
||
before = len(fake.questions)
|
||
resp = client.post("/api/search", json=body)
|
||
data = resp.json() if resp.status_code == 200 else {}
|
||
ok = resp.status_code == 200 and data.get("code") == 1
|
||
record(f"server 接受 OCS payload：{name}", ok, f"HTTP {resp.status_code} {str(data)[:120]}")
|
||
if len(fake.questions) > before:
|
||
q = fake.questions[-1]
|
||
# server.py 对 title/options/suggestion_* 做了 .strip()（首尾空白），
|
||
# 标记 " [图片N] " 在正文中间，不会被破坏。
|
||
record(
|
||
f"server 正确透传新字段：{name}",
|
||
q.images == body.get("images")
|
||
and q.suggestion_title == (body.get("suggestion_title") or "").strip()
|
||
and q.suggestion_options == (body.get("suggestion_options") or "").strip()
|
||
and q.title == (body.get("title") or "").strip()
|
||
and q.options == (body.get("options") or "").strip(),
|
||
f"images={len(q.images)} sug_title={q.suggestion_title[:40]!r}",
|
||
)
|
||
|
||
# 3.2 字段异常/缺失的容器行为（server 只做容器归一化，不做语义修正）
|
||
abnormal = [
|
||
(
|
||
"旧版 payload（无 images/suggestion）",
|
||
{"title": "2 + 3 = ?", "type": "single", "options": "A. 4\nB. 5"},
|
||
lambda q: q.images == [] and q.suggestion_title == "" and q.suggestion_options == "",
|
||
),
|
||
(
|
||
"images 不是 list（字符串）",
|
||
{"title": "t", "type": "single", "options": "A. 1\nB. 2", "images": "data:image/png;base64,AAA"},
|
||
lambda q: q.images == [],
|
||
),
|
||
(
|
||
"images 不是 list（dict）",
|
||
{"title": "t", "type": "single", "options": "A. 1\nB. 2", "images": {"0": "x"}},
|
||
lambda q: q.images == [],
|
||
),
|
||
(
|
||
"images 含非字符串/空值",
|
||
{"title": "t", "type": "single", "options": "A. 1\nB. 2",
|
||
"images": [None, "", 123, {}, question_png]},
|
||
lambda q: isinstance(q.images, list) and len(q.images) == 5,
|
||
),
|
||
(
|
||
"suggestion_* 为数字/None",
|
||
{"title": "t", "type": "single", "options": "A. 1\nB. 2",
|
||
"images": [question_png], "suggestion_title": 123, "suggestion_options": None},
|
||
lambda q: q.suggestion_title == "123" and q.suggestion_options == "",
|
||
),
|
||
(
|
||
"suggestion_* 为 'undefined' 字面量（旧 OCS 可能残留）",
|
||
{"title": "t", "type": "single", "options": "A. 1\nB. 2",
|
||
"images": [], "suggestion_title": "undefined", "suggestion_options": "undefined"},
|
||
lambda q: q.suggestion_title == "undefined" and q.suggestion_options == "undefined",
|
||
),
|
||
]
|
||
for name, payload, predicate in abnormal:
|
||
before = len(fake.questions)
|
||
resp = client.post("/api/search", json=payload)
|
||
data = resp.json() if resp.status_code == 200 else {}
|
||
ok_http = resp.status_code == 200 and data.get("code") == 1
|
||
ok = ok_http and len(fake.questions) > before and predicate(fake.questions[-1])
|
||
record(f"server 字段容错：{name}", ok,
|
||
f"HTTP {resp.status_code} code={data.get('code')}")
|
||
|
||
# 3.3 ai_client 对 server 透传的异常值继续兜底
|
||
from ai_client import _clean_ocs_text, _normalize_ocs_image, _prepare_ocs_images
|
||
|
||
record("ai_client._clean_ocs_text 把 'undefined'/'null'/非字符串当空",
|
||
_clean_ocs_text("undefined") == "" and _clean_ocs_text("null") == ""
|
||
and _clean_ocs_text(None) == "" and _clean_ocs_text(123) == "")
|
||
record("ai_client._normalize_ocs_image 过滤空/非字符串/非 data URL",
|
||
_normalize_ocs_image(None) == "" and _normalize_ocs_image("") == ""
|
||
and _normalize_ocs_image("undefined") == "" and _normalize_ocs_image(123) == ""
|
||
and _normalize_ocs_image(question_png) == question_png)
|
||
filtered = _prepare_ocs_images([None, "", "undefined", 123, {}, question_png])
|
||
record("ai_client._prepare_ocs_images 过滤后保留有效图", filtered == [question_png],
|
||
f"len={len(filtered)}")
|
||
|
||
# 3.4 日志脱敏：base64 不得进入日志
|
||
capture = LogCapture()
|
||
ocs_logger = logging.getLogger("ocs_ai")
|
||
ocs_logger.addHandler(capture)
|
||
try:
|
||
big = "data:image/png;base64," + "A" * 200_000
|
||
capture.reset()
|
||
resp = client.post(
|
||
"/api/search",
|
||
json={"title": "日志脱敏测试", "type": "single", "options": "A. 1\nB. 2",
|
||
"images": [big], "suggestion_title": " [图片1] ", "suggestion_options": ""},
|
||
)
|
||
messages = list(capture.messages)
|
||
finally:
|
||
ocs_logger.removeHandler(capture)
|
||
|
||
joined = "\n".join(messages)
|
||
body_logs = [m for m in messages if "POST 请求体" in m]
|
||
record(
|
||
"日志脱敏：base64 图片不进日志",
|
||
"AAAAAAAA" not in joined and "base64," not in joined.split("POST 请求体", 1)[-1][:200],
|
||
f"maxlogline={max((len(m) for m in messages), default=0)}",
|
||
)
|
||
record(
|
||
"日志脱敏：images 字段显示为 '<N 张 base64 图片>'",
|
||
any("<1 张 base64 图片>" in m for m in body_logs),
|
||
str(body_logs[:1])[:200],
|
||
)
|
||
record(
|
||
"日志脱敏：单条日志长度 < 5000 字符",
|
||
max((len(m) for m in messages), default=0) < 5000,
|
||
f"max={max((len(m) for m in messages), default=0)}",
|
||
)
|
||
|
||
# 3.5 GET 接口（旧客户端）仍可用
|
||
before = len(fake.questions)
|
||
resp = client.get("/api/search", params={"title": "1 + 1 = ?", "type": "single",
|
||
"options": "A. 2\nB. 3"})
|
||
data = resp.json() if resp.status_code == 200 else {}
|
||
record("旧版 GET /api/search 仍兼容", resp.status_code == 200 and data.get("code") == 1)
|
||
except Exception as exc:
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
record("server 字段解析测试执行", False, repr(exc))
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 四、OCS 源码静态断言（确认 emulator 移植与源码一致）
|
||
# ══════════════════════════════════════════════════════════════════
|
||
def section_source() -> None:
|
||
print("\n[四] OCS 5.0 源码静态核对（确认判定条件 / payload 格式）")
|
||
|
||
src = find_ocs_src()
|
||
if src is None:
|
||
skip("OCS 源码静态核对", "未找到源码目录（可用 OCS5_SRC 环境变量指定）")
|
||
return
|
||
|
||
env_ts_path = src / "packages/scripts/src/utils/answerer-env.ts"
|
||
handler_ts_path = src / "packages/core/src/core/answer-wrapper/answer.wrapper.handler.ts"
|
||
common_ts_path = src / "packages/scripts/src/projects/common.ts"
|
||
parser_ts_path = src / "packages/core/src/core/answer-wrapper/answer.wrapper.parser.ts"
|
||
request_ts_path = src / "packages/core/src/core/utils/request.ts"
|
||
|
||
texts = {}
|
||
for key, path in [
|
||
("env", env_ts_path),
|
||
("handler", handler_ts_path),
|
||
("common", common_ts_path),
|
||
("parser", parser_ts_path),
|
||
("request", request_ts_path),
|
||
]:
|
||
if path.is_file():
|
||
texts[key] = path.read_text(encoding="utf-8")
|
||
|
||
# 4.1 检测逻辑关键条件
|
||
env = texts.get("env", "")
|
||
record("源码：检测要求 method 为 post",
|
||
"String(wrapper.method || 'get').toLowerCase() !== 'post'" in env)
|
||
record("源码：检测三个图片占位符",
|
||
all(p in env for p in ('`\\${images}`', '`\\${suggestion_title}`', '`\\${suggestion_options}`')))
|
||
record("源码：data 字段 handler 的 env 引用正则",
|
||
"IMAGE_ENV_REF_PATTERN = /(^|[^\\w$])(images|suggestion_title|suggestion_options)(?![\\w$])/" in env)
|
||
record("源码：仅检测 data 字段值（字符串占位符 / 对象 handler）",
|
||
"typeof value === 'string'" in env and "typeof (value as any).handler === 'string'" in env)
|
||
|
||
# 4.2 env 构建：无图时不注入 images/suggestion_*
|
||
record("源码：仅当 createImageSuggestion 有结果才注入 env.images",
|
||
"if (suggestion) {" in env and "env.images = suggestion.images" in env)
|
||
record("源码：suggestion_options 仅在有选项图片时设置",
|
||
"if (suggestion.suggestion_options !== undefined)" in env)
|
||
|
||
# 4.3 占位符格式：' [图片N] '，左右各一个空格
|
||
line = ""
|
||
for raw_line in env.splitlines():
|
||
if "placeholderMap.set(url" in raw_line:
|
||
line = raw_line.strip()
|
||
break
|
||
record("源码：占位符模板为 ' [图片${images.length + 1}] '（左右各一空格）",
|
||
" [图片${images.length + 1}] " in line, line)
|
||
# 用源码里的模板生成真实标记，喂给中间件解析器
|
||
real_marker = " [图片7] "
|
||
from ai_client import _iter_marker_segments
|
||
|
||
segs = _iter_marker_segments("A" + real_marker + "B")
|
||
record("中间件能识别源码格式的标记（含空格）",
|
||
[(k, v) for k, v in segs if k == "image"] == [("image", 7)], str(segs))
|
||
|
||
# 4.4 运行时无图时 env 默认值（避免 undefined 字面量）
|
||
handler = texts.get("handler", "")
|
||
record("源码：无图时 env.images 默认 []",
|
||
"if (env.images === undefined) env.images = [];" in handler)
|
||
record("源码：无图时 suggestion_title 默认 ''",
|
||
"if (env.suggestion_title === undefined) env.suggestion_title = '';" in handler)
|
||
record("源码：无图时 suggestion_options 默认 ''",
|
||
"if (env.suggestion_options === undefined) env.suggestion_options = '';" in handler)
|
||
|
||
# 4.5 POST data 解析：data.* handler 用 Function 执行，字符串走占位符替换
|
||
record("源码：POST data 字段 handler 由 Function 动态执行",
|
||
"Function(Reflect.get((wrapperData as any)[key], 'handler'))()" in handler)
|
||
record("源码：POST data 字符串字段走 resolvePlaceHolder",
|
||
"Reflect.set(data, key, resolvePlaceHolder(wrapperData[key]));" in handler)
|
||
record("源码：POST 请求体最终 JSON.stringify(data)",
|
||
"JSON.stringify(data)" in texts.get("request", ""))
|
||
|
||
# 4.6 警告逻辑与开关默认值
|
||
common = texts.get("common", "")
|
||
record("源码：警告仅在「图片题优化开启 + 有题库 + 配置不支持」时出现",
|
||
"isAnswererWrappersSupportImageOptimize(wrappers)" in common
|
||
and "当前题库配置暂不支持图片、公式等题目处理" in common)
|
||
record("源码：图片题优化开关默认开启",
|
||
re.search(r"imageOptimize:\s*\{.*?defaultValue:\s*true", common, re.S) is not None)
|
||
|
||
# 4.7 配置解析：粘贴字符串只做 JSON.parse（不强制 type 字段），fromObject 需要 name/url/handler
|
||
parser = texts.get("parser", "")
|
||
record("源码：粘贴 JSON 字符串只走 JSON.parse",
|
||
"return this.fromJSONString(value);" in parser)
|
||
record("源码：fromObject 必需字段 name/url/handler",
|
||
all(f"typeof item.{k} !== 'string'" in parser for k in ("name", "url", "handler")))
|
||
record("源码：type 字段缺失不会导致校验失败（仅当存在时校验取值）",
|
||
"if (item.type && types.every((i) => i !== item.type))" in parser)
|
||
|
||
# 4.8 与 daemon 行为相关的补充说明（打印给报告用）
|
||
print(" [INFO] 关键结论：")
|
||
print(" - 判定 = method 必须 post + data 值里有 ${...} 占位符 或 data.*.handler 引用 env 图片字段")
|
||
print(" - 运行时 env 固定含 title/options/type；无图时 images=[]、suggestion_*=''")
|
||
print(" - 有图时 images 元素统一为 data:image/png;base64,...，标记格式 ' [图片N] '（N 从 1 起，先 title 后 options）")
|
||
print(" - suggestion_options 仅在选项含图片时出现（缺失时 handler 会补空字符串）")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
def main() -> int:
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
print("=" * 74)
|
||
print("OCS 5.0「图片题优化」契约测试（离线，不调用 DeepSeek）")
|
||
print("=" * 74)
|
||
node_ok, node = node_available()
|
||
print(f"Node.js: {node or '未找到'}")
|
||
print(f"OCS 源码: {find_ocs_src() or '未找到'}")
|
||
|
||
generated = section_config()
|
||
bodies = section_payload(generated)
|
||
section_server(bodies)
|
||
section_source()
|
||
|
||
print("\n" + "=" * 74)
|
||
if FAILS:
|
||
print(f"总体: FAIL ❌ ({len(FAILS)} 项失败)")
|
||
for name in FAILS:
|
||
print(f" - {name}")
|
||
if SKIPS:
|
||
print(f"跳过 {len(SKIPS)} 项（环境原因）")
|
||
print("=" * 74)
|
||
return 1
|
||
print(f"总体: PASS ✅" + (f"（跳过 {len(SKIPS)} 项）" if SKIPS else ""))
|
||
print("=" * 74)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|