cat-shark
dcdc5e8604
批量引擎改为「分块流水线」：视频按 WOV_BATCH_STAGE_GROUP_SIZE（默认 8）分组， 组内按 DAG 拓扑序跑完全部视频（全部 extract → 全部 ASR → 全部翻译 → 全部 ASS） 再进入下一组，本地模型每组只加载一次、卸载一次，而不是每个视频来回加载卸载； 产物仍按组增量落到视频旁。调度器新增 execute_run(run_id, stop_after=节点)： 该节点完成后任务保持 RUNNING 不收尾，下一次调用从产物表跳过已完成节点继续， 用于实现阶段边界。 - nodes/llm.py：翻译节点结束释放本机 Ollama 显存（node 参数 unload_after > LLM_UNLOAD_AFTER > 本机 loopback 端点默认卸载，云端端点不卸载；卸载失败只告警）， 新增 keep_model.flag 语义（阶段内保持常驻）与 release_local_model()； 新增节点内暂停（按批 20 行检查 paused.flag，抛 PauseRequested，调度器保持 PAUSED）。 - src/wov_app/batch.py：分组阶段执行与阶段末统一释放显存；失败视频只在它失败 节点的那个阶段重试（避免 LLM 已常驻时重跑 ASR 抢显存）；任务没有明细时保持 QUEUED 等登记完成、仍有未完成视频时置回 QUEUED 自愈（原先留 RUNNING 会卡死： 引擎只拾取 QUEUED，任务停在“运行中但没人推进”）；无失败视频时删除任务级空目录； 每个阶段开始前清理 paused.flag / keep_model.flag，避免强杀残留影响后续阶段。 - src/wov_app/config.py：新增 WOV_BATCH_STAGE_GROUP_SIZE（设为 1 即旧的每视频全链路）。 - 任务列表与批量页分工：GET /api/runs 默认排除 source=batch（一个批量任务会产生 N 条单视频 run，会把 20 条窗口占满；且任务管理页的暂停/重试/删除对批量 run 语义不成立），需要排查时用 include_batch=1；作为补偿批量页详情新增阶段列 （阶段 i/N · 中文标签，由该视频 run 的 current_node_id 在 DAG 拓扑序中的位置 推导，节点类型映射中文标签）。阶段只有节点边界粒度，句级进度不落库、只在日志。 - 顺带纳入此前未提交的批量僵尸状态恢复：recover_interrupted_batch_jobs 除 RUNNING 外也把「COMPLETED 但仍含未结束视频」的任务置回 QUEUED；fix_zombie_batch_jobs.py 改为按条件扫描并支持 --apply 预览；批量页明细只列本批真正处理过的视频。 测试新增/更新：分块流水线调用顺序（组内按节点跑完再下一组）、每组只释放一次模型、 阶段内保持常驻标志、翻译按批暂停、失败视频不跨阶段推进、任务无明细/中途登记视频时 置回 QUEUED、任务工作空间与残留信号清理、任务列表默认过滤批量 run、详情阶段字段、 前端阶段列渲染；全量 507 passed（唯一失败为既有素材缺失的 integration 用例）。
740 lines
27 KiB
Python
740 lines
27 KiB
Python
"""nodes/llm.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/llm.py`（LLM 翻译节点：分批请求 + 按 ID 回填），可独立调用。
|
||
网络属于允许 mock 的 I/O 边界：单元用例注入假 HTTP 响应验证请求体、ID 校验、
|
||
重试与回填；集成用例调用真实 LLM 验证真实字幕翻译质量。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from nodes.llm import (
|
||
CHUNK_SIZE,
|
||
MAX_BATCH_RETRIES,
|
||
PauseRequested,
|
||
_parse_translations,
|
||
_system_prompt,
|
||
invoke,
|
||
release_local_model,
|
||
translate_lines,
|
||
)
|
||
from wov_sdk.models import InvokeRequest
|
||
|
||
# 模块专用数据目录（真实字幕产物缺失时相关用例跳过）。
|
||
DATA_DIR = Path(__file__).resolve().parent / "data"
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate_llm_env(monkeypatch):
|
||
"""清掉外部泄漏的 LLM 路由变量，保证用例只受自己显式设置的环境变量影响。
|
||
|
||
全量跑时其它模块 import 应用会触发 load_dotenv()，把开发者 .env 里的
|
||
LLM_API_BASE（可能指向本机 Ollama）带进来，从而改变端点判定与请求数量。
|
||
"""
|
||
for name in ("LLM_API_BASE", "LLM_MODEL", "LLM_UNLOAD_AFTER"):
|
||
monkeypatch.delenv(name, raising=False)
|
||
|
||
|
||
class _FakeHTTPResponse:
|
||
"""假的 HTTP 响应：返回预置 JSON 体（供 urlopen mock 使用）。"""
|
||
|
||
def __init__(self, payload: dict) -> None:
|
||
self._data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
|
||
def read(self) -> bytes:
|
||
return self._data
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc) -> None:
|
||
return None
|
||
|
||
|
||
def _llm_reply(translations: list[tuple[int, str]], total_tokens: int = 42) -> _FakeHTTPResponse:
|
||
"""构造 OpenAI 兼容接口的响应体（content 为 {id,text} JSON 数组）。"""
|
||
content = json.dumps(
|
||
[{"id": i, "text": t} for i, t in translations], ensure_ascii=False
|
||
)
|
||
return _FakeHTTPResponse({
|
||
"choices": [{"message": {"content": content}}],
|
||
"usage": {"total_tokens": total_tokens},
|
||
})
|
||
|
||
|
||
def _capture_urlopen(calls: list[dict], responses: list[_FakeHTTPResponse]):
|
||
"""返回一个假 urlopen：记录请求体，按顺序返回预置响应。"""
|
||
|
||
def fake_urlopen(http_request, timeout=None):
|
||
calls.append({
|
||
"url": http_request.full_url,
|
||
"body": json.loads(http_request.data.decode("utf-8")),
|
||
"headers": dict(http_request.headers),
|
||
"timeout": timeout,
|
||
})
|
||
return responses.pop(0) if responses else _llm_reply([])
|
||
|
||
return fake_urlopen
|
||
|
||
|
||
def _capture_urlopen_routing_unload(
|
||
calls: list[dict],
|
||
responses: list[_FakeHTTPResponse],
|
||
unload_error: Exception | None = None,
|
||
):
|
||
"""按 URL 分流的假 urlopen：卸载请求只记录，翻译请求按序返回预置响应。"""
|
||
|
||
def fake_urlopen(http_request, timeout=None):
|
||
calls.append({
|
||
"url": http_request.full_url,
|
||
"body": json.loads(http_request.data.decode("utf-8")),
|
||
"timeout": timeout,
|
||
})
|
||
if http_request.full_url.endswith("/api/generate"):
|
||
if unload_error is not None:
|
||
raise unload_error
|
||
return _FakeHTTPResponse({"done": True})
|
||
return responses.pop(0) if responses else _llm_reply([])
|
||
|
||
return fake_urlopen
|
||
|
||
|
||
def _local_llm_env(monkeypatch) -> None:
|
||
"""把 LLM 端点指向本地 Ollama（qwen3:30b-a3b）。"""
|
||
monkeypatch.setenv("LLM_API_KEY", "")
|
||
monkeypatch.setenv("LLM_API_BASE", "http://localhost:11434/v1/chat/completions")
|
||
monkeypatch.setenv("LLM_MODEL", "qwen3:30b-a3b")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 提示词与响应解析（纯函数）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_system_prompt_mentions_target_language_and_json_contract() -> None:
|
||
"""系统提示词说明目标语言与 JSON 条目契约（时间戳不进入模型）。"""
|
||
# 数据：目标语言 zh-CN。
|
||
# 测试过程
|
||
prompt = _system_prompt("zh-CN")
|
||
|
||
# 验证结果
|
||
assert "zh-CN" in prompt
|
||
assert "id" in prompt and "text" in prompt
|
||
|
||
|
||
def test_parse_translations_accepts_out_of_order_ids() -> None:
|
||
"""乱序返回的条目按 ID 回填（不依赖数组顺序）。"""
|
||
# 数据：ID 为 3、1、2 的乱序结果。
|
||
content = json.dumps([
|
||
{"id": 3, "text": "三"},
|
||
{"id": 1, "text": "一"},
|
||
{"id": 2, "text": "二"},
|
||
])
|
||
|
||
# 测试过程
|
||
parsed = _parse_translations(content, {1, 2, 3})
|
||
|
||
# 验证结果
|
||
assert parsed == {1: "一", 2: "二", 3: "三"}
|
||
|
||
|
||
def test_parse_translations_rejects_missing_id() -> None:
|
||
"""缺少任一 ID 时明确报错（不允许静默漏译导致时间轴错位）。"""
|
||
# 数据：缺少 ID 2。
|
||
content = json.dumps([{"id": 1, "text": "一"}, {"id": 3, "text": "三"}])
|
||
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="missing"):
|
||
_parse_translations(content, {1, 2, 3})
|
||
|
||
|
||
def test_parse_translations_rejects_duplicate_id() -> None:
|
||
"""重复 ID 报错。"""
|
||
# 数据：ID 1 出现两次。
|
||
content = json.dumps([{"id": 1, "text": "一"}, {"id": 1, "text": "壹"}])
|
||
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="duplicate"):
|
||
_parse_translations(content, {1})
|
||
|
||
|
||
def test_parse_translations_rejects_empty_text() -> None:
|
||
"""空正文或非字符串正文报错。"""
|
||
# 数据：空字符串与数字正文。
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="empty or invalid"):
|
||
_parse_translations(json.dumps([{"id": 1, "text": " "}]), {1})
|
||
with pytest.raises(ValueError, match="empty or invalid"):
|
||
_parse_translations(json.dumps([{"id": 1, "text": 3}]), {1})
|
||
|
||
|
||
def test_parse_translations_rejects_unexpected_id() -> None:
|
||
"""返回了未请求的 ID 时报错。"""
|
||
# 数据：包含 ID 9（未请求）。
|
||
content = json.dumps([{"id": 9, "text": "九"}])
|
||
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="invalid or duplicate"):
|
||
_parse_translations(content, {1})
|
||
|
||
|
||
def test_parse_translations_rejects_non_array_payload() -> None:
|
||
"""顶层不是数组时报错。"""
|
||
# 数据：对象形式的返回。
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="JSON array"):
|
||
_parse_translations(json.dumps({"id": 1, "text": "一"}), {1})
|
||
|
||
|
||
def test_parse_translations_keeps_multiline_text_structure() -> None:
|
||
"""译文多行结构保留（去除纯空行，不截断 cue）。"""
|
||
# 数据：含空行的多行译文。
|
||
content = json.dumps([{"id": 1, "text": "第一行\n\n第二行"}])
|
||
|
||
# 测试过程
|
||
parsed = _parse_translations(content, {1})
|
||
|
||
# 验证结果：空行被去掉但两行都保留。
|
||
assert parsed[1] == "第一行\n第二行"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# translate_lines：请求体、ID 与重试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_translate_lines_sends_global_ids_and_maps_back(monkeypatch) -> None:
|
||
"""按全局位置 ID 请求翻译，并把结果按位置回填（空 cue 不请求但占位）。"""
|
||
# 数据：5 行，其中第 3 行为空（占位）。
|
||
lines = ["一", "二", "", "四", "五"]
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "one"), (2, "two"), (4, "four"), (5, "five")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
result = translate_lines(lines, {"target_language": "en"})
|
||
|
||
# 验证结果：请求体只含非空行的全局 ID（1,2,4,5），结果按位置回填且空行保留。
|
||
sent = json.loads(calls[0]["body"]["messages"][1]["content"])
|
||
assert [item["id"] for item in sent] == [1, 2, 4, 5]
|
||
assert result == ["one", "two", "", "four", "five"]
|
||
|
||
|
||
def test_translate_lines_retries_on_structure_error(monkeypatch) -> None:
|
||
"""结构校验失败时重试，最终成功（最多 MAX_BATCH_RETRIES 次）。"""
|
||
# 数据：第一次返回缺 ID，第二次正确。
|
||
lines = ["一", "二"]
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [
|
||
_llm_reply([(1, "one")]), # 缺 ID 2 → 触发重试
|
||
_llm_reply([(1, "one"), (2, "two")]), # 正确
|
||
]),
|
||
)
|
||
|
||
# 测试过程
|
||
result = translate_lines(lines, {})
|
||
|
||
# 验证结果：重试一次后成功，共发起 2 次请求。
|
||
assert result == ["one", "two"]
|
||
assert len(calls) == 2
|
||
|
||
|
||
def test_translate_lines_raises_after_retries_exhausted(monkeypatch) -> None:
|
||
"""结构错误耗尽重试后抛错（不返回错位译文）。"""
|
||
# 数据：每次都返回错误结构。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([]) for _ in range(MAX_BATCH_RETRIES)]),
|
||
)
|
||
|
||
# 测试过程与验证结果
|
||
with pytest.raises(ValueError, match="alignment failed"):
|
||
translate_lines(["一"], {})
|
||
assert len(calls) == MAX_BATCH_RETRIES
|
||
|
||
|
||
def test_translate_lines_empty_input_makes_no_request(monkeypatch) -> None:
|
||
"""空输入直接返回空列表，不发请求。"""
|
||
# 数据：空列表。
|
||
calls: list[dict] = []
|
||
monkeypatch.setattr(urllib.request, "urlopen", _capture_urlopen(calls, []))
|
||
|
||
# 测试过程与验证结果
|
||
assert translate_lines([], {}) == []
|
||
assert calls == []
|
||
|
||
|
||
def test_translate_lines_all_empty_cues_make_no_request(monkeypatch) -> None:
|
||
"""全部为空 cue 时不请求模型，返回等长空列表。"""
|
||
# 数据：3 个空行。
|
||
calls: list[dict] = []
|
||
monkeypatch.setattr(urllib.request, "urlopen", _capture_urlopen(calls, []))
|
||
|
||
# 测试过程与验证结果
|
||
assert translate_lines(["", " ", ""], {}) == ["", "", ""]
|
||
assert calls == []
|
||
|
||
|
||
def test_translate_lines_batches_by_chunk_size(monkeypatch) -> None:
|
||
"""超过 CHUNK_SIZE 行时分批请求（每批最多 CHUNK_SIZE 条）。"""
|
||
# 数据：CHUNK_SIZE + 1 行。
|
||
total = CHUNK_SIZE + 1
|
||
lines = [f"行{i}" for i in range(1, total + 1)]
|
||
calls: list[dict] = []
|
||
|
||
def fake_urlopen(http_request, timeout=None):
|
||
body = json.loads(http_request.data.decode("utf-8"))
|
||
calls.append(body)
|
||
items = json.loads(body["messages"][1]["content"])
|
||
return _llm_reply([(item["id"], f"t{item['id']}") for item in items])
|
||
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
||
|
||
# 测试过程
|
||
result = translate_lines(lines, {})
|
||
|
||
# 验证结果：两批（CHUNK_SIZE + 1），结果等长且顺序正确。
|
||
assert len(calls) == 2
|
||
assert len(result) == total
|
||
assert result[0] == "t1" and result[-1] == f"t{total}"
|
||
|
||
|
||
def test_translate_lines_injects_proper_noun_rule(monkeypatch) -> None:
|
||
"""本批原文命中专名时，系统提示词追加规则（不被硬译）。"""
|
||
# 数据：含专名 ジンゴ 的一行。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["ジンゴがここで味わいませんか"], {})
|
||
|
||
# 验证结果：系统提示词包含该专名与禁止硬译的说明。
|
||
system_prompt = calls[0]["body"]["messages"][0]["content"]
|
||
assert "ジンゴ" in system_prompt
|
||
assert "芒果" in system_prompt
|
||
|
||
|
||
def test_translate_lines_default_model_and_env_override(monkeypatch) -> None:
|
||
"""默认模型来自 LLM_MODEL 环境变量（数据驱动，不改代码切换模型）。"""
|
||
# 数据：设置环境变量为自定义模型。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_MODEL", "自定义/模型")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "x")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果：请求体使用环境变量指定的模型。
|
||
assert calls[0]["body"]["model"] == "自定义/模型"
|
||
|
||
|
||
def test_translate_lines_param_model_wins_over_env(monkeypatch) -> None:
|
||
"""节点参数 model 优先于环境变量。"""
|
||
# 数据：环境变量与参数都设置。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_MODEL", "env/模型")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "x")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {"model": "param/模型"})
|
||
|
||
# 验证结果
|
||
assert calls[0]["body"]["model"] == "param/模型"
|
||
|
||
|
||
def test_translate_lines_uses_timeout_env(monkeypatch) -> None:
|
||
"""LLM_TIMEOUT_SECONDS 决定请求超时（默认 600）。"""
|
||
# 数据：设置 45 秒。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_TIMEOUT_SECONDS", "45")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "x")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果
|
||
assert calls[0]["timeout"] == 45.0
|
||
|
||
|
||
def test_translate_lines_sends_bearer_key(monkeypatch) -> None:
|
||
"""请求头带 Bearer Key（来自 LLM_API_KEY）。"""
|
||
# 数据：设置 key。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-abc")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "x")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果
|
||
headers = {k.lower(): v for k, v in calls[0]["headers"].items()}
|
||
assert headers.get("authorization") == "Bearer sk-abc"
|
||
|
||
|
||
def test_translate_lines_unloads_local_model_when_env_enabled(monkeypatch) -> None:
|
||
"""开启 LLM_UNLOAD_AFTER 时翻译结束请求 Ollama 卸载模型，把显存让给 whisper。"""
|
||
# 数据：本地端点 + 开启卸载。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.setenv("LLM_UNLOAD_AFTER", "1")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果：翻译后向 Ollama 原生端点发 keep_alive=0 的卸载请求。
|
||
assert [call["url"] for call in calls] == [
|
||
"http://localhost:11434/v1/chat/completions",
|
||
"http://localhost:11434/api/generate",
|
||
]
|
||
assert calls[1]["body"] == {"model": "qwen3:30b-a3b", "keep_alive": 0}
|
||
|
||
|
||
def test_translate_lines_auto_unloads_loopback_endpoint(monkeypatch) -> None:
|
||
"""端点在本机（loopback）时默认卸载：无需开关，默认就让出显存。"""
|
||
# 数据：本地端点 + 不设置任何开关。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.delenv("LLM_UNLOAD_AFTER", raising=False)
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果：本机端点默认发出卸载请求。
|
||
assert calls[-1]["url"] == "http://localhost:11434/api/generate"
|
||
|
||
|
||
def test_translate_lines_does_not_unload_remote_endpoint(monkeypatch) -> None:
|
||
"""云端端点默认不卸载：显存不由本机持有，多发请求只是噪声。"""
|
||
# 数据：远程端点 + 不设置任何开关。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions")
|
||
monkeypatch.delenv("LLM_UNLOAD_AFTER", raising=False)
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {})
|
||
|
||
# 验证结果：只有翻译请求。
|
||
assert [call["url"] for call in calls] == ["https://api.siliconflow.cn/v1/chat/completions"]
|
||
|
||
|
||
def test_translate_lines_param_unload_after_enables_unload(monkeypatch) -> None:
|
||
"""节点参数 unload_after=True 等效于环境变量开关。"""
|
||
# 数据：只给节点参数。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.delenv("LLM_UNLOAD_AFTER", raising=False)
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {"unload_after": True})
|
||
|
||
# 验证结果
|
||
assert calls[-1]["url"] == "http://localhost:11434/api/generate"
|
||
|
||
|
||
def test_translate_lines_param_unload_after_false_overrides_default(monkeypatch) -> None:
|
||
"""节点参数 unload_after=False 可显式关闭本机端点的默认卸载。"""
|
||
# 数据：本机端点 + 节点参数显式关闭。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.setenv("LLM_UNLOAD_AFTER", "1")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {"unload_after": False})
|
||
|
||
# 验证结果
|
||
assert [call["url"] for call in calls] == ["http://localhost:11434/v1/chat/completions"]
|
||
|
||
|
||
def test_translate_lines_keeps_translation_when_unload_fails(monkeypatch) -> None:
|
||
"""卸载请求失败不影响译文（端点不支持卸载时只是跳过释放）。"""
|
||
# 数据：远程端点显式开启卸载 + 卸载请求返回 404。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_UNLOAD_AFTER", "1")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(
|
||
calls,
|
||
[_llm_reply([(1, "译文")])],
|
||
unload_error=urllib.error.HTTPError(
|
||
"https://api.example.com/api/generate", 404, "not found", {}, None
|
||
),
|
||
),
|
||
)
|
||
|
||
# 测试过程
|
||
translated = translate_lines(["一"], {})
|
||
|
||
# 验证结果：译文正常返回，卸载失败只记录。
|
||
assert translated == ["译文"]
|
||
assert calls[-1]["url"].endswith("/api/generate")
|
||
|
||
|
||
def test_translate_lines_keeps_model_loaded_in_staged_batch(monkeypatch) -> None:
|
||
"""引擎标记阶段内保持常驻时不卸载模型：整批只加载一次，阶段结束统一释放。"""
|
||
# 数据：本机端点（默认会卸载）+ keep_model_loaded=True。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
|
||
# 测试过程
|
||
translate_lines(["一"], {}, keep_model_loaded=True)
|
||
|
||
# 验证结果：只发翻译请求，不发卸载请求。
|
||
assert [call["url"] for call in calls] == ["http://localhost:11434/v1/chat/completions"]
|
||
|
||
|
||
def test_translate_lines_aborts_between_batches_when_stop_requested(monkeypatch) -> None:
|
||
"""暂停信号在两批之间生效：已完成的批保留，后续批不再发请求。"""
|
||
# 数据：CHUNK_SIZE + 1 行（两批），第二次检查返回“应停止”。
|
||
lines = [f"行{i}" for i in range(1, CHUNK_SIZE + 2)]
|
||
calls: list[dict] = []
|
||
checks = {"count": 0}
|
||
|
||
def stop_requested() -> bool:
|
||
checks["count"] += 1
|
||
return checks["count"] > 1
|
||
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(
|
||
calls,
|
||
[_llm_reply([(i, f"t{i}") for i in range(1, CHUNK_SIZE + 1)])],
|
||
),
|
||
)
|
||
|
||
# 测试过程与验证结果：抛暂停异常，且只发出第一批的请求。
|
||
with pytest.raises(PauseRequested):
|
||
translate_lines(lines, {}, stop_requested=stop_requested)
|
||
assert [call["url"] for call in calls] == ["https://api.siliconflow.cn/v1/chat/completions"]
|
||
|
||
|
||
def test_release_local_model_unloads_loopback_endpoint(monkeypatch) -> None:
|
||
"""阶段收尾释放模型：本机端点发 keep_alive=0 卸载请求，模型名参数优先。"""
|
||
# 数据：本机端点 + 显式指定的模型名。
|
||
calls: list[dict] = []
|
||
_local_llm_env(monkeypatch)
|
||
monkeypatch.setattr(urllib.request, "urlopen", _capture_urlopen_routing_unload(calls, []))
|
||
|
||
# 测试过程
|
||
release_local_model("local/替换模型")
|
||
|
||
# 验证结果：命中 Ollama 原生卸载端点，使用传入的模型名。
|
||
assert [(call["url"], call["body"]) for call in calls] == [
|
||
("http://localhost:11434/api/generate", {"model": "local/替换模型", "keep_alive": 0}),
|
||
]
|
||
|
||
|
||
def test_release_local_model_skips_remote_endpoint(monkeypatch) -> None:
|
||
"""云端端点不占本机显存，阶段收尾不发卸载请求。"""
|
||
# 数据：云端端点。
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions")
|
||
monkeypatch.setattr(urllib.request, "urlopen", _capture_urlopen_routing_unload(calls, []))
|
||
|
||
# 测试过程
|
||
release_local_model()
|
||
|
||
# 验证结果：没有任何请求。
|
||
assert calls == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# invoke 全流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_invoke_translates_srt_and_writes_artifact(monkeypatch, tmp_path: Path) -> None:
|
||
"""invoke 解析 SRT → 翻译 → 写出 cn_srt，时间轴保持原样。"""
|
||
# 数据：两条真实 SRT。
|
||
srt = "1\n00:00:01,000 --> 00:00:02,000\nこんにちは\n\n2\n00:00:03,000 --> 00:00:04,000\nさようなら\n"
|
||
srt_path = tmp_path / "in.srt"
|
||
srt_path.write_text(srt, encoding="utf-8")
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen(calls, [_llm_reply([(1, "你好"), (2, "再见")])]),
|
||
)
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={},
|
||
inputs={"srt_uri": str(srt_path)}, output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果：状态、产物时间轴与译文顺序。
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["cn_srt_uri"]).read_text(encoding="utf-8")
|
||
assert "00:00:01,000 --> 00:00:02,000" in content
|
||
assert "你好" in content and "再见" in content
|
||
assert content.index("你好") < content.index("再见")
|
||
|
||
|
||
def test_invoke_stops_on_pause_flag(monkeypatch, tmp_path: Path) -> None:
|
||
"""run 根目录有暂停信号时节点中止且不写产物（调度器保持任务 PAUSED）。"""
|
||
# 数据：一条真实 SRT + run 根目录下的 paused.flag。
|
||
srt_path = tmp_path / "in.srt"
|
||
srt_path.write_text("1\n00:00:01,000 --> 00:00:02,000\nこんにちは\n", encoding="utf-8")
|
||
run_root = tmp_path / "runs" / "run-paused"
|
||
run_root.mkdir(parents=True, exist_ok=True)
|
||
(run_root / "paused.flag").write_text("", encoding="utf-8")
|
||
output_dir = run_root / "steps" / "translate"
|
||
calls: list[dict] = []
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
monkeypatch.setattr(
|
||
urllib.request, "urlopen",
|
||
_capture_urlopen_routing_unload(calls, [_llm_reply([(1, "译文")])]),
|
||
)
|
||
request = InvokeRequest(
|
||
run_id="run-paused", node_instance_id="n", params={},
|
||
inputs={"srt_uri": str(srt_path)}, output_dir=str(output_dir),
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果：failed 且原因为暂停；未调用 LLM、未写 cn.srt。
|
||
assert response.status == "failed"
|
||
assert "暂停" in (response.error or "")
|
||
assert calls == []
|
||
assert not (output_dir / "cn.srt").exists()
|
||
|
||
|
||
def test_invoke_fails_without_input(tmp_path: Path) -> None:
|
||
"""缺少 srt_uri 时失败。"""
|
||
# 数据：空输入。
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={}, inputs={}, output_dir=str(tmp_path)
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "srt_uri" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_input_missing(tmp_path: Path) -> None:
|
||
"""输入文件不存在时失败。"""
|
||
# 数据：不存在的路径。
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={},
|
||
inputs={"srt_uri": str(tmp_path / "nope.srt")}, output_dir=str(tmp_path),
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "not found" in (response.error or "")
|
||
|
||
|
||
def test_invoke_reports_failure_on_llm_error(monkeypatch, tmp_path: Path) -> None:
|
||
"""LLM 报错时节点返回 failed（不产出半成品产物）。"""
|
||
# 数据：urlopen 抛 HTTPError。
|
||
srt_path = tmp_path / "in.srt"
|
||
srt_path.write_text("1\n00:00:01,000 --> 00:00:02,000\nこんにちは\n", encoding="utf-8")
|
||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||
|
||
def fake_urlopen(http_request, timeout=None):
|
||
raise urllib.error.HTTPError(http_request.full_url, 500, "server error", {}, None)
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={},
|
||
inputs={"srt_uri": str(srt_path)}, output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 真实 LLM 集成（需要 LLM_API_KEY 与网络）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
def test_real_llm_translates_short_lines_with_domain_rules() -> None:
|
||
"""真实 LLM 校准：短句专名/隐语不按字面直译（真实 API，外部状态缺失则跳过）。"""
|
||
# 数据：含专名 ジンゴ 的短句；Key 与账号可用性由共享判定处理。
|
||
from nodes.llm import translate_lines as real_translate
|
||
from tests.shared.llm_service import require_llm_credentials, skip_on_service_unavailable
|
||
|
||
require_llm_credentials()
|
||
|
||
# 测试过程
|
||
with skip_on_service_unavailable():
|
||
out = real_translate(["ジンゴがここで味わいませんか"], {"target_language": "zh-CN"})
|
||
|
||
# 验证结果：有译文且未把专名硬译成"芒果"。
|
||
assert out and out[0].strip()
|
||
assert "芒果" not in out[0]
|