3218485270
b413104587
- 豆包: response_format.json_schema(strict)；DeepSeek: 切 Responses API text.format.json_schema （官方 chat/completions 通道不支持 json_schema），不支持时自动降级 json_object（[output] schema） - 重试两段式：传输错误 cfg.retries 次；解析失败/空正文有独立 3 次专用重试，仍失败落 parse_fail - 移除三票复核（decide_final/_verify_clean/verify_clean），DeepSeek 单次判定即终稿 - 新增 [run] deepseek_recheck 开关（默认 no=仅豆包初筛；yes=追加 DeepSeek 复检） - GUI 不再覆盖 config 提示词；prompt 相对/前导斜杠路径按 exe 目录解析 - 报表去「票型/复核」列并同步说明；README/config.example.ini 同步 - 测试新增输出格式、重试、开关与提示词路径用例（50 passed）
270 lines
10 KiB
Python
270 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""providers 模块测试：判定 JSON 解析、分类归一化、图片编码、usage 规整、output_schema、重试。"""
|
||
import asyncio
|
||
import base64
|
||
|
||
import pytest
|
||
|
||
from violation_detector.config import AppConfig
|
||
from violation_detector.providers import (
|
||
CANON, OUTPUT_SCHEMA, SCHEMA_CATEGORIES, SCHEMA_NAME, _chat_detect,
|
||
_extract_output_text, _norm_usage, _result_responses,
|
||
_schema_explicitly_unsupported, build_chat_payload,
|
||
build_responses_payload, encode_image, norm_category, parse_judgment,
|
||
)
|
||
|
||
GOOD = ('{"文字/图形属性": "黑色T恤印字母", "侵权/违规逻辑": "未授权使用商标", '
|
||
'"违规分类": "12. 侵权 - 除人物外的其他侵权"}')
|
||
|
||
|
||
def test_parse_plain_json():
|
||
r = parse_judgment(GOOD)
|
||
assert r is not None
|
||
assert r[0] == "黑色T恤印字母"
|
||
assert r[1] == "未授权使用商标"
|
||
assert r[2] == "12. 侵权 - 除人物外的其他侵权"
|
||
|
||
|
||
def test_parse_fenced_json():
|
||
r = parse_judgment("```json\n" + GOOD + "\n```")
|
||
assert r is not None and r[2].startswith("12.")
|
||
|
||
|
||
def test_parse_json_with_surrounding_text():
|
||
r = parse_judgment("分析如下：\n" + GOOD + "\n以上。")
|
||
assert r is not None
|
||
|
||
|
||
def test_parse_invalid():
|
||
assert parse_judgment("") is None
|
||
assert parse_judgment("这不是JSON") is None
|
||
assert parse_judgment('{"文字/图形属性": "x"}') is None # 缺字段
|
||
assert parse_judgment('{"a": [1,2]') is None # 截断
|
||
|
||
|
||
def test_norm_category_variants():
|
||
# 不同写法归一到标准 17 类名称
|
||
assert norm_category("12. 侵权 - 人物外") == CANON[12]
|
||
assert norm_category("14脏话、侮辱性") == CANON[14]
|
||
assert norm_category("5. 负向敏感") == CANON[5]
|
||
# 非编号类原样保留
|
||
assert norm_category("无违规") == "无违规"
|
||
assert norm_category("违规不明") == "违规不明"
|
||
# 未知编号保留原文
|
||
assert norm_category("99. 未来分类") == "99. 未来分类"
|
||
|
||
|
||
def test_encode_image_roundtrip(tmp_path):
|
||
img = tmp_path / "a.jpg"
|
||
img.write_bytes(b"\xff\xd8\xff\xe0fake")
|
||
url = encode_image(str(img))
|
||
assert url.startswith("data:image/jpeg;base64,")
|
||
assert base64.b64decode(url.split(",", 1)[1]) == b"\xff\xd8\xff\xe0fake"
|
||
|
||
|
||
def test_encode_image_unsupported(tmp_path):
|
||
p = tmp_path / "a.txt"
|
||
p.write_text("x", encoding="utf-8")
|
||
with pytest.raises(ValueError):
|
||
encode_image(str(p))
|
||
|
||
|
||
def test_norm_usage_handles_missing_details():
|
||
assert _norm_usage({"prompt_tokens": 100, "completion_tokens": 50}) == {
|
||
"prompt": 100, "completion": 50, "cached": 0}
|
||
assert _norm_usage({}) == {"prompt": None, "completion": None, "cached": 0}
|
||
assert _norm_usage({"prompt_tokens": 100, "prompt_tokens_details": {"cached_tokens": 80}})["cached"] == 80
|
||
|
||
|
||
# ---------- output_schema ----------
|
||
|
||
def test_output_schema_shape():
|
||
assert OUTPUT_SCHEMA["type"] == "object"
|
||
assert set(OUTPUT_SCHEMA["required"]) == {"文字/图形属性", "侵权/违规逻辑", "违规分类"}
|
||
assert OUTPUT_SCHEMA["additionalProperties"] is False
|
||
assert set(OUTPUT_SCHEMA["properties"]) == set(OUTPUT_SCHEMA["required"])
|
||
|
||
|
||
def test_schema_categories_covers_canon_and_specials():
|
||
expect = [CANON[n] for n in sorted(CANON)] + ["无违规", "违规不明"]
|
||
assert SCHEMA_CATEGORIES == expect
|
||
|
||
|
||
def test_build_chat_payload_schema_vs_json():
|
||
# 豆包：结构化输出 = response_format.json_schema(strict)；降级 = json_object
|
||
messages = [{"role": "user", "content": []}]
|
||
s = build_chat_payload("m", messages, {"temperature": 0}, True)
|
||
assert s["model"] == "m" and s["temperature"] == 0
|
||
rf = s["response_format"]
|
||
assert rf["type"] == "json_schema"
|
||
assert rf["json_schema"]["name"] == SCHEMA_NAME
|
||
assert rf["json_schema"]["strict"] is True
|
||
assert rf["json_schema"]["schema"] is OUTPUT_SCHEMA
|
||
j = build_chat_payload("m", messages, {}, False)
|
||
assert j["response_format"] == {"type": "json_object"}
|
||
|
||
|
||
def test_build_responses_payload_schema():
|
||
# DeepSeek：结构化输出 = text.format.json_schema（Responses API）
|
||
items = [{"role": "user", "content": [{"type": "input_text", "text": "p"}]}]
|
||
p = build_responses_payload("m", items, {"max_output_tokens": 5000}, True)
|
||
assert p["model"] == "m" and p["max_output_tokens"] == 5000
|
||
fmt = p["text"]["format"]
|
||
assert fmt["type"] == "json_schema"
|
||
assert fmt["name"] == SCHEMA_NAME
|
||
assert fmt["schema"] is OUTPUT_SCHEMA
|
||
|
||
|
||
def test_extract_output_text_concats_output_text():
|
||
data = {"output": [
|
||
{"type": "reasoning", "role": "assistant",
|
||
"content": [{"type": "reasoning_text", "text": "思考"}],
|
||
"status": "completed"},
|
||
{"type": "message", "role": "assistant",
|
||
"content": [{"type": "output_text", "text": '{"a": 1}'}],
|
||
"status": "completed"},
|
||
]}
|
||
assert _extract_output_text(data) == '{"a": 1}'
|
||
|
||
|
||
def test_result_responses_ok():
|
||
data = {
|
||
"status": "completed",
|
||
"usage": {"input_tokens": 10, "output_tokens": 20,
|
||
"input_tokens_details": {"cached_tokens": 4}},
|
||
"output": [{"type": "message", "role": "assistant", "content": [
|
||
{"type": "output_text", "text": '{"文字/图形属性": "T恤", '
|
||
'"侵权/违规逻辑": "无", "违规分类": "无违规"}'}]}],
|
||
}
|
||
r = _result_responses("a.jpg", data)
|
||
assert r["status"] == "ok" and r["category"] == "无违规"
|
||
assert r["usage"] == {"prompt": 10, "completion": 20, "cached": 4}
|
||
assert r["finish_reason"] == "completed"
|
||
|
||
|
||
def test_schema_explicitly_unsupported():
|
||
assert _schema_explicitly_unsupported(
|
||
"Unrecognized request argument supplied: response_format.json_schema")
|
||
assert _schema_explicitly_unsupported("model not support json_schema")
|
||
assert _schema_explicitly_unsupported("不支持的参数 json_schema")
|
||
assert not _schema_explicitly_unsupported("HTTP 400 some other problem")
|
||
|
||
|
||
# ---------- 重试机制（错误码 / 空正文 / 解析失败） ----------
|
||
|
||
RETRY_GOOD = '{"文字/图形属性": "T恤", "侵权/违规逻辑": "无", "违规分类": "无违规"}'
|
||
RETRY_BAD = '{"文字/图形属性": "T恤"}' # 缺字段，解析失败
|
||
|
||
|
||
def _chat_ok(content):
|
||
return {"choices": [{"message": {"content": content}, "finish_reason": "stop"}],
|
||
"usage": {}}
|
||
|
||
|
||
class _FakeResp:
|
||
def __init__(self, status, data=None, body=""):
|
||
self.status = status
|
||
self._data = data
|
||
self._body = body
|
||
|
||
async def json(self):
|
||
return self._data
|
||
|
||
async def text(self):
|
||
return self._body
|
||
|
||
|
||
class _FakeCM:
|
||
def __init__(self, resp):
|
||
self._resp = resp
|
||
|
||
async def __aenter__(self):
|
||
return self._resp
|
||
|
||
async def __aexit__(self, *exc):
|
||
return False
|
||
|
||
|
||
class _FakeSession:
|
||
def __init__(self, responses):
|
||
self.responses = list(responses)
|
||
self.posts = 0
|
||
|
||
def post(self, url, json, headers):
|
||
resp = self.responses[min(self.posts, len(self.responses) - 1)]
|
||
self.posts += 1
|
||
return _FakeCM(resp)
|
||
|
||
|
||
def _run_chat(session, retries=3):
|
||
cfg = AppConfig(output_schema_mode="off", retries=retries)
|
||
return asyncio.run(_chat_detect(
|
||
session, asyncio.Semaphore(1), cfg, "a.jpg", stop=lambda: False,
|
||
provider="deepseek", label="DeepSeek",
|
||
base_url="https://fake.deepseek.com", api_key="k", model="m",
|
||
messages=[{"role": "user", "content": "p"}],
|
||
extras={"temperature": 0, "max_tokens": 50}))
|
||
|
||
|
||
async def _no_sleep(_):
|
||
return None
|
||
|
||
|
||
def test_parse_fail_retries_then_recovers(monkeypatch):
|
||
# 第 1 次返回不可解析 → 触发重试；第 2 次成功
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD)),
|
||
_FakeResp(200, _chat_ok(RETRY_GOOD))])
|
||
res = _run_chat(sess)
|
||
assert res["status"] == "ok" and res["category"] == "无违规"
|
||
assert sess.posts == 2
|
||
|
||
|
||
def test_parse_fail_exhausts_to_parse_fail(monkeypatch):
|
||
# 一直解析失败 → 第 1 次进入解析段，独立再重试 3 次后落 parse_fail（违规不明）
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD))])
|
||
res = _run_chat(sess, retries=3)
|
||
assert res["status"] == "parse_fail" and res["category"] == "违规不明"
|
||
assert sess.posts == 1 + 3 # 首判 + 3 次独立解析重试
|
||
|
||
|
||
def test_parse_fail_dedicated_budget_ignores_http_errors(monkeypatch):
|
||
# 先 1 次 HTTP 500，再解析失败：HTTP 错误不消耗解析段次数，
|
||
# 解析仍独立获得 cfg.retries 次专用重试
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(500, body="boom"),
|
||
_FakeResp(200, _chat_ok(RETRY_BAD))])
|
||
res = _run_chat(sess, retries=3)
|
||
assert res["status"] == "parse_fail" and res["category"] == "违规不明"
|
||
assert sess.posts == 2 + 3 # 1×HTTP500 + 首判解析失败 + 3×解析专用重试
|
||
|
||
|
||
def test_error_code_retries_then_recovers(monkeypatch):
|
||
# HTTP 500 → 退避重试；随后 200 成功
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(500, body="boom"),
|
||
_FakeResp(200, _chat_ok(RETRY_GOOD))])
|
||
res = _run_chat(sess)
|
||
assert res["status"] == "ok" and res["category"] == "无违规"
|
||
assert sess.posts == 2
|
||
|
||
|
||
def test_error_code_exhausts_to_error(monkeypatch):
|
||
# 一直 HTTP 500 → 重试耗尽落 error（检测异常）
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(500, body="boom")])
|
||
res = _run_chat(sess, retries=3)
|
||
assert res["status"] == "error" and res["category"] == "检测异常"
|
||
assert sess.posts == 3
|
||
|
||
|
||
def test_empty_content_retries(monkeypatch):
|
||
# 空正文 → 重试；随后成功
|
||
monkeypatch.setattr(asyncio, "sleep", _no_sleep)
|
||
sess = _FakeSession([_FakeResp(200, _chat_ok("")),
|
||
_FakeResp(200, _chat_ok(RETRY_GOOD))])
|
||
res = _run_chat(sess)
|
||
assert res["status"] == "ok" and res["category"] == "无违规"
|
||
assert sess.posts == 2
|