Blame for tests/test_service.py

"""任务状态/隔离回归测试，以及真实 FFmpeg 合成测试。

Blender 调用在这里由测试替身替代；这些测试不表示 Blender 或 ZeroGPU
已通过实测。最后的视频测试实际运行 FFmpeg/FFprobe 并读取生成的 MP4。
"""

from __future__ import annotations

import importlib
import json
import shutil
import subprocess
import sys
import threading
import time
import zipfile
from pathlib import Path

import pytest
from PIL import Image

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))


@pytest.fixture
def service(monkeypatch, tmp_path):
    monkeypatch.setenv("RENDER_DATA_DIR", str(tmp_path / "runtime"))
    monkeypatch.setenv("RENDER_BACKEND", "cpu")
    monkeypatch.setenv("MIN_FREE_MB", "1")
    module = importlib.import_module("service")
    data = tmp_path / "runtime"
    for attribute, value in (("DATA_ROOT", data), ("JOB_ROOT", data / "jobs"), ("PUBLIC_ROOT", data / "public")):
        value.mkdir(parents=True, exist_ok=True)
        monkeypatch.setattr(module, attribute, value)
    monkeypatch.setattr(module, "_LOCKS", {})
    monkeypatch.setattr(module, "ensure_blender", lambda: Path("/test/fake-blender"))
    monkeypatch.setattr(module, "_backend", lambda cfg=None: "cpu")
    return module


@pytest.fixture
def make_job(service, tmp_path):
    def create(**config):
        source = tmp_path / "scene.blend"
        source.write_bytes(b"TEST ONLY: Blender is not invoked by these unit tests")
        settings = {"mode": "frames", "width": 32, "height": 24, "device": "CPU",
                    "samples": 1, "frame_start": 1, "frame_end": 5, "batch_size": 2}
        settings.update(config)
        return service.prepare_job([str(source)], json.dumps(settings))
    return create


def write_frames(task: dict, frames: list[int] | None = None) -> None:
    """使用真实 PNG 文件模拟 Blender 已原子提交的帧输出。"""
    from render_config import RenderConfig
    config = RenderConfig.model_validate(task["config"])
    output = Path(task["output_dir"])
    output.mkdir(parents=True, exist_ok=True)
    for frame in frames if frames is not None else task["frames"]:
        file = output / f"frame_{frame:06d}.png"
        Image.new("RGB", (config.output_width, config.output_height),
                  (frame * 17 % 256, frame * 31 % 256, frame * 47 % 256)).save(file)


@pytest.fixture
def successful_worker(service, monkeypatch):
    calls = []

    def run(task, timeout, log_path, cancel_path=None):
        calls.append(task)
        Path(log_path).write_text("Unit-test worker; no GPU or Blender execution.\n")
        if task["action"] == "inspect":
            return {"ok": True, "scenes": ["Scene"], "cameras": ["Camera"]}
        write_frames(task)
        return {"ok": True, "backend": "CPU", "frames": task["frames"]}

    monkeypatch.setattr(service, "run_task", run)
    return calls


def test_prepare_inspect_render_in_batches_and_idempotent_complete(service, make_job, successful_worker):
    job, status = make_job()
    assert len(job) == 48 and status["status"] == "prepared"
    assert status["completed_count"] == 0 and status["total_frames"] == 5
    assert status["config"]["device"] == "CPU"
    assert service.inspect_job(job)["status"] == "ready"
    first = service.render_chunk(job)
    assert first["status"] == "partial"
    assert first["completed_frames"] == [1, 2]
    assert first["remaining_frames"] == [3, 4, 5]
    assert first["progress"] == 0.4
    service.render_chunk(job)
    final = service.render_chunk(job)
    assert final["status"] == "complete" and final["progress"] == 1
    assert [call["frames"] for call in successful_worker if call["action"] == "render"] == [[1, 2], [3, 4], [5]]
    count = len(successful_worker)
    assert service.render_chunk(job)["status"] == "complete"
    assert len(successful_worker) == count


def test_prepare_config_file_takes_precedence_and_accepts_bom(service, tmp_path):
    blend = tmp_path / "scene.blend"
    blend.write_bytes(b"test")
    config_file = tmp_path / "render.json"
    config_file.write_text('{"width":64,"height":32}', encoding="utf-8-sig")
    _, status = service.prepare_job([str(blend)], "invalid unused text", str(config_file))
    assert status["config"]["width"] == 64
    config_file.write_text("x" * 65537)
    before = set(service.JOB_ROOT.iterdir())
    with pytest.raises(ValueError, match="64"):
        service.prepare_job([str(blend)], "{}", str(config_file))
    assert set(service.JOB_ROOT.iterdir()) == before


def test_prepare_failure_leaves_no_job_and_rejects_capacity(service, tmp_path, make_job, monkeypatch):
    script = tmp_path / "bad.py"
    script.write_text("raise RuntimeError('never execute')")
    with pytest.raises(ValueError):
        service.prepare_job([str(script)], "{}")
    assert list(service.JOB_ROOT.iterdir()) == []
    monkeypatch.setenv("MAX_JOBS", "1")
    make_job()
    with pytest.raises(ValueError, match="上限"):
        make_job()
    assert len(list(service.JOB_ROOT.iterdir())) == 1


@pytest.mark.parametrize("job_id", ["../secret", "/tmp/file", "", "x" * 48, "0" * 48])
def test_private_job_ids_cannot_be_paths_or_unknown_jobs(service, job_id):
    with pytest.raises(ValueError):
        service.job_status(job_id)


def test_job_files_and_cancellation_are_isolated(service, make_job, successful_worker):
    first, _ = make_job()
    second, _ = make_job()
    assert first != second
    service.render_chunk(first)
    assert service.job_status(second)["completed_frames"] == []
    service.cancel_job(first)
    service.render_chunk(second)
    assert not (service.JOB_ROOT / second / "cancel.flag").exists()
    summary, files, image, video = service.finalize_job(second)
    assert image and not video
    assert all(Path(file).parent == service.PUBLIC_ROOT / second for file in files)
    assert summary["job_id"] == second
    archive = next(Path(file) for file in files if file.endswith(".zip"))
    with zipfile.ZipFile(archive) as data:
        names = data.namelist()
        assert "manifest.json" in names and "frames/frame_000001.png" in names
        assert not any(name.endswith(".blend") or name.startswith("project/") for name in names)
        assert json.loads(data.read("manifest.json"))["job_id"] == second


def test_partial_failure_retains_committed_frame_and_retries_only_missing(service, make_job, monkeypatch):
    job, _ = make_job(batch_size=3)
    seen = []

    def failing_once(task, timeout, log_path, cancel_path=None):
        seen.append(task["frames"])
        if len(seen) == 1:
            write_frames(task, task["frames"][:1])
            raise TimeoutError("Synthetic worker timeout after one committed frame")
        write_frames(task)
        return {"ok": True}

    monkeypatch.setattr(service, "run_task", failing_once)
    partial = service.render_chunk(job)
    assert partial["status"] == "partial" and partial["completed_frames"] == [1]
    assert "Synthetic worker timeout" in partial["error"]
    next_status = service.render_chunk(job)
    assert seen == [[1, 2, 3], [2, 3, 4]]
    assert next_status["error"] is None
    assert service.render_chunk(job)["status"] == "complete"


def test_worker_error_result_is_failed_and_not_success(service, make_job, monkeypatch):
    job, _ = make_job()
    monkeypatch.setattr(service, "run_task", lambda *a, **k: {"ok": False, "error": "Synthetic render failure"})
    status = service.render_chunk(job)
    assert status["status"] == "failed"
    assert status["completed_frames"] == [] and "Synthetic render failure" in status["error"]


def test_inspection_error_result_does_not_mark_scene_ready(service, make_job, monkeypatch):
    job, _ = make_job()
    monkeypatch.setattr(service, "run_task", lambda *a, **k: {"ok": False, "error": "Synthetic missing scene"})
    status = service.inspect_job(job)
    assert status["status"] == "failed"
    assert "Synthetic missing scene" in status["error"]


def test_idle_cancel_is_visible_and_resume_retains_frames(service, make_job, successful_worker):
    job, _ = make_job()
    service.render_chunk(job)
    paused = service.cancel_job(job)
    assert paused["status"] == "cancelled"
    with pytest.raises(ValueError, match="暂停|取消"):
        service.render_chunk(job)
    resumed = service.resume_job(job)
    assert resumed["completed_frames"] == [1, 2]
    assert not (service.JOB_ROOT / job / "cancel.flag").exists()
    service.render_chunk(job)
    assert [call["frames"] for call in successful_worker] == [[1, 2], [3, 4]]


def test_active_cancel_and_lock_protect_running_batch(service, make_job, monkeypatch):
    job, _ = make_job()
    started, release = threading.Event(), threading.Event()
    response = {}

    def waiting_worker(task, timeout, log_path, cancel_path=None):
        write_frames(task, task["frames"][:1])
        started.set()
        if not release.wait(5):
            raise TimeoutError("Test synchronization timed out")
        assert Path(cancel_path).exists()
        raise RuntimeError("Synthetic cancellation observed")

    monkeypatch.setattr(service, "run_task", waiting_worker)

    def render():
        try:
            response["status"] = service.render_chunk(job)
        except BaseException as exc:
            response["exception"] = exc

    thread = threading.Thread(target=render, daemon=True)
    thread.start()
    try:
        assert started.wait(5)
        assert service.job_status(job)["status"] == "rendering"
        with pytest.raises(ValueError, match="执行"):
            service.render_chunk(job)
        with pytest.raises(ValueError, match="执行"):
            service.resume_job(job)
        assert service.cancel_job(job)["status"] == "cancelling"
    finally:
        release.set()
        thread.join(timeout=5)
    assert not thread.is_alive()
    assert "exception" not in response
    assert response["status"]["status"] == "cancelled"
    assert response["status"]["completed_frames"] == [1]


def test_corrupt_and_wrong_size_frames_are_rerendered(service, make_job, successful_worker):
    job, _ = make_job(batch_size=5)
    frames = service.JOB_ROOT / job / "frames"
    (frames / "frame_000001.png").write_bytes(b"corrupt PNG" * 30)
    Image.new("RGB", (16, 16)).save(frames / "frame_000002.png")
    (frames / "frame_000003.part.png").write_bytes(b"unfinished" * 30)
    status = service.render_chunk(job)
    assert status["status"] == "complete"
    assert successful_worker[0]["frames"] == [1, 2, 3, 4, 5]


def test_final_manifest_uses_verified_progress_consistently(service, make_job):
    job, _ = make_job()
    path = service.JOB_ROOT / job
    config = service._load(path)
    write_frames(service._task(path, config, "render", [1]))
    (path / "frames/frame_000002.png").write_bytes(b"corrupt PNG" * 30)
    summary, files, _, _ = service.finalize_job(job)
    assert summary["completed_frames"] == [1]
    assert summary["completed_count"] == 1
    assert summary["remaining_frames"] == [2, 3, 4, 5]
    assert summary["progress"] == 0.2
    assert summary["export_complete"] is False


def test_resume_reports_corrupt_frames_as_pending(service, make_job):
    job, _ = make_job()
    path = service.JOB_ROOT / job
    meta = service._load(path)
    write_frames(service._task(path, meta, "render", [1]))
    (path / "frames/frame_000002.png").write_bytes(b"corrupt PNG" * 30)
    service.cancel_job(job)
    summary = service.resume_job(job)
    assert summary["completed_frames"] == [1]
    assert summary["completed_count"] == 1
    assert summary["remaining_frames"] == [2, 3, 4, 5]
    assert summary["progress"] == 0.2


def test_file_lock_open_failure_does_not_permanently_lock_job(service, make_job, monkeypatch):
    job, _ = make_job()
    path = service.JOB_ROOT / job
    original_open = Path.open
    state = {"fail": True}

    def fail_lock_once(self, *args, **kwargs):
        if self.name == ".lock" and state["fail"]:
            state["fail"] = False
            raise OSError("Synthetic filesystem failure")
        return original_open(self, *args, **kwargs)

    monkeypatch.setattr(Path, "open", fail_lock_once)
    with pytest.raises(OSError, match="Synthetic"):
        with service._lock(path):
            pass
    with service._lock(path):
        pass


def test_expiration_does_not_delete_active_locked_job(service, make_job, monkeypatch):
    monkeypatch.setenv("JOB_TTL_HOURS", "1")
    old_job, _ = make_job()
    active_job, _ = make_job()
    for job in (old_job, active_job):
        path = service.JOB_ROOT / job
        meta = service._load(path)
        meta["updated_at"] = time.time() - 7200
        service._atomic_json(path / "job.json", meta)
        (service.PUBLIC_ROOT / job).mkdir()
    with service._lock(service.JOB_ROOT / active_job):
        service._cleanup()
        assert (service.JOB_ROOT / active_job).exists()
        assert (service.PUBLIC_ROOT / active_job).exists()
    assert not (service.JOB_ROOT / old_job).exists()
    assert not (service.PUBLIC_ROOT / old_job).exists()


@pytest.mark.skipif(not shutil.which("ffmpeg") or not shutil.which("ffprobe"), reason="真实视频验证需要 FFmpeg 和 FFprobe")
def test_real_ffmpeg_video_preserves_frame_step_timing_and_scaled_dimensions(service, make_job, successful_worker):
    job, _ = make_job(mode="video", width=64, height=48, resolution_percentage=50,
                      frame_start=1, frame_end=9, frame_step=2, fps=24, batch_size=5)
    before, _, _, video = service.finalize_job(job)
    assert video is None and before["export_complete"] is False
    assert service.render_chunk(job)["status"] == "complete"
    summary, files, image, video = service.finalize_job(job)
    assert summary["export_complete"] is True and image and video
    assert summary["video_playback_fps"] == 12
    result = subprocess.run(
        [shutil.which("ffprobe"), "-v", "error", "-select_streams", "v:0", "-show_entries",
         "stream=codec_name,width,height,nb_frames,avg_frame_rate,duration", "-of", "json", video],
        check=True, capture_output=True, text=True, timeout=30,
    )
    stream = json.loads(result.stdout)["streams"][0]
    assert stream["codec_name"] == "h264"
    assert (stream["width"], stream["height"]) == (32, 24)
    assert int(stream["nb_frames"]) == 5
    numerator, denominator = map(int, stream["avg_frame_rate"].split("/"))
    assert numerator / denominator == pytest.approx(12)
    assert float(stream["duration"]) == pytest.approx(5 / 12, abs=0.02)
    assert not (service.JOB_ROOT / job / "video-sequence").exists()
    archive = next(file for file in files if file.endswith(".zip"))
    with zipfile.ZipFile(archive) as exported:
        assert "animation.mp4" in exported.namelist()
        assert len([name for name in exported.namelist() if name.startswith("frames/")]) == 5


def test_video_encoder_failure_keeps_frame_export(service, make_job, successful_worker, monkeypatch):
    job, _ = make_job(mode="video", frame_end=1)
    service.render_chunk(job)

    def unavailable(path, cfg):
        raise RuntimeError("Synthetic FFmpeg failure")

    monkeypatch.setattr(service, "_encode_video", unavailable)
    summary, files, image, video = service.finalize_job(job)
    assert "Synthetic FFmpeg failure" in summary["export_error"]
    assert summary["export_complete"] is False and image and video is None
    assert any(file.endswith(".zip") for file in files)


def test_private_frame_paths_do_not_become_gradio_downloads(service):
    """真实回归：Client 会自动下载 JSON 内的 path，导致私有帧路径 403。"""
    from gradio_client import utils
    report = service._public_report({
        'frames': [{'frame': 1, 'path': '/private/job/frames/frame_000001.png'}],
        'ok': True,
    })
    found = []
    utils.traverse(report, lambda obj: found.append(obj) or obj, utils.is_file_obj)
    assert not found
    assert report['frames'][0]['filename'] == 'frame_000001.png'