67 lines
2.9 KiB
Python
67 lines
2.9 KiB
Python
"""R06：真实图片清单上的临时网络故障、空帧与跨空白字幕段回归。"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nodes import subtitle_ocr
|
|
from wov_sdk.models import InvokeRequest, InvokeResponse
|
|
|
|
|
|
def test_identical_text_separated_by_blank_is_two_cues():
|
|
"""A、空白、A 不能合并，否则字幕会覆盖原本无文字的时段。"""
|
|
manifest = [{"time": i * 0.5} for i in range(4)]
|
|
assert subtitle_ocr._merge_kept(manifest, ["你好", "你好", "", "你好"]) == [
|
|
(0.0, 0.5, "你好"), (1.5, 1.5, "你好")]
|
|
|
|
|
|
@pytest.mark.parametrize("raises", [False, True])
|
|
def test_failed_frame_retries_and_resume_preserves_success(monkeypatch, tmp_path, raises):
|
|
"""失败帧单独重试，持续故障不存为空；下次只补失败帧，成功空帧不重做。"""
|
|
assets = Path(__file__).resolve().parent.parent / "testdata"
|
|
image = assets / "ocr_text.png"
|
|
empty = assets / "ocr_notext.png"
|
|
if not image.is_file() or not empty.is_file():
|
|
pytest.skip("缺少真实 OCR 图片")
|
|
manifest = tmp_path / "frames.json"
|
|
manifest.write_text(json.dumps([{"time": 0, "image_uri": str(empty)},
|
|
{"time": 0.5, "image_uri": str(image)}]))
|
|
out = tmp_path / "out"
|
|
request = InvokeRequest(run_id="r", node_instance_id="", inputs={"frames_manifest": str(manifest)}, output_dir=str(out))
|
|
calls = []
|
|
broken = True
|
|
|
|
def invoke(node_id, req):
|
|
uri = req.inputs["image_uri"]
|
|
calls.append(uri)
|
|
if uri == str(image) and broken:
|
|
if raises:
|
|
raise TimeoutError("timeout")
|
|
return InvokeResponse(status="failed", error="timeout")
|
|
return InvokeResponse(status="completed", outputs={"text": "" if uri == str(empty) else "你好"})
|
|
|
|
monkeypatch.setattr("wov_app.registry.invoke", invoke)
|
|
response = subtitle_ocr.invoke(request)
|
|
assert response.status == "failed"
|
|
assert calls.count(str(image)) == 2
|
|
assert calls.count(str(empty)) == 1
|
|
partial = [json.loads(line) for line in (out / "ocr_partial.jsonl").read_text().splitlines()]
|
|
assert partial == [{"frame": 0, "text": "", "status": "completed"}]
|
|
assert not (out / "subtitle.srt").exists()
|
|
broken = False
|
|
calls.clear()
|
|
response = subtitle_ocr.invoke(request)
|
|
assert response.status == "completed"
|
|
assert calls == [str(image)]
|
|
assert "你好" in Path(response.outputs["srt_uri"]).read_text()
|
|
|
|
|
|
def test_legacy_empty_checkpoint_is_rechecked(tmp_path):
|
|
"""旧版空串可能来自超时，不能当成确认无文字；旧版非空成功结果可复用。"""
|
|
(tmp_path / "ocr_partial.jsonl").write_text(
|
|
json.dumps({"frame": 0, "text": ""}) + "\n" +
|
|
json.dumps({"frame": 1, "text": "你好"}, ensure_ascii=False) + "\n" +
|
|
json.dumps({"frame": 2, "text": "", "status": "completed"}) + "\n")
|
|
assert subtitle_ocr._load_partial(tmp_path) == {1: "你好", 2: ""}
|