xiaoxia-saas/tests/unit/test_oneclick_gen_p0_fixes.py

"""P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验.

覆盖：
  P0-2: OSS 上传失败抛异常 + URL 可访问性校验
  P0-3: FFmpeg 失败时完整 stderr 日志
  P1:   template_id 存在性校验 + asset_ids 归属校验
"""

from __future__ import annotations

import logging
import subprocess
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# 添加 worker app 到 sys.path
_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
if str(_WORKER_ROOT) not in sys.path:
    sys.path.insert(0, str(_WORKER_ROOT))


# ── P0-1: _download_library_assets ────────────────────────────────────────────


class TestOSSUploadAndVerify:
    """P0-2: OSS 上传失败抛异常 + URL 可访问性校验."""

    def test_verify_url_accessible_success(self):
        """URL 可访问时返回 True."""
        from worker_app.tasks.generation import _verify_url_accessible

        mock_response = MagicMock()
        mock_response.status = 200
        mock_response.__enter__ = MagicMock(return_value=mock_response)
        mock_response.__exit__ = MagicMock(return_value=False)

        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
            assert _verify_url_accessible("https://example.com/test.mp4") is True

    def test_verify_url_accessible_failure(self):
        """URL 不可访问时返回 False."""
        from worker_app.tasks.generation import _verify_url_accessible

        with patch("urllib.request.OpenerDirector.open", side_effect=Exception("connection refused")):
            assert _verify_url_accessible("https://example.com/test.mp4") is False

    def test_verify_url_404(self):
        """URL 返回 404 时返回 False."""
        from worker_app.tasks.generation import _verify_url_accessible

        mock_response = MagicMock()
        mock_response.status = 404
        mock_response.__enter__ = MagicMock(return_value=mock_response)
        mock_response.__exit__ = MagicMock(return_value=False)

        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
            assert _verify_url_accessible("https://example.com/test.mp4") is False


# ── P0-3: FFmpeg stderr 日志 ─────────────────────────────────────────────────


class TestFFmpegStderrLogging:
    """P0-3: FFmpeg 失败时完整 stderr 打到日志."""

    def test_run_ffmpeg_logs_stderr_on_failure(self, caplog):
        """run_ffmpeg 失败时记录 stderr 到日志."""
        from video_processing.ffmpeg_utils import run_ffmpeg

        error = subprocess.CalledProcessError(
            returncode=183,
            cmd=["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"],
            output="",
            stderr="Error message from ffmpeg: filter graph error details here",
        )

        with patch("subprocess.run", side_effect=error):
            with caplog.at_level(logging.ERROR):
                with pytest.raises(subprocess.CalledProcessError):
                    run_ffmpeg(["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"])

        assert "FFmpeg 命令失败" in caplog.text
        assert "exit_code=183" in caplog.text
        assert "filter graph error" in caplog.text

    def test_run_ffmpeg_success(self):
        """run_ffmpeg 成功时正常返回."""
        from video_processing.ffmpeg_utils import run_ffmpeg

        mock_result = MagicMock()
        mock_result.stdout = "output"
        mock_result.stderr = ""

        with patch("subprocess.run", return_value=mock_result):
            stdout, stderr = run_ffmpeg(["ffmpeg", "-version"])
            assert stdout == "output"
            assert stderr == ""


# ── P1: template_id + asset_ids 校验 ─────────────────────────────────────────


class TestP1Validations:
    """P1: template_id 存在性校验 + asset_ids 归属校验."""

    def test_validate_template_exists_success(self):
        """模板存在时不抛异常."""
        from worker_app.tasks.generation import _validate_template_exists

        mock_template = MagicMock()
        mock_template.id = "tmpl_001"
        mock_template.name = "Test Template"
        mock_template.is_active = True
        mock_template.status = "active"
        mock_template.version = 1

        session = MagicMock()

        # EditTemplateModel 查询返回 None（走旧模板系统 fallback）
        edit_query = MagicMock()
        edit_filter = MagicMock()
        edit_query.filter.return_value = edit_filter
        edit_filter.first.return_value = None

        # TemplateModel 查询返回 mock_template
        old_query = MagicMock()
        old_filter = MagicMock()
        old_query.filter.return_value = old_filter
        old_filter.first.return_value = mock_template

        def _query_side_effect(model):
            # 根据 model 类型返回不同的 query mock
            name = getattr(model, "__name__", "")
            if "EditTemplate" in name:
                return edit_query
            return old_query

        session.query.side_effect = _query_side_effect

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            _validate_template_exists("tmpl_001")  # 不抛异常

    def test_validate_template_exists_not_found(self):
        """模板不存在时抛 ValueError."""
        from worker_app.tasks.generation import _validate_template_exists

        session = MagicMock()

        # 两个系统查询都返回 None
        def _query_side_effect(model):
            q = MagicMock()
            f = MagicMock()
            q.filter.return_value = f
            f.first.return_value = None
            return q

        session.query.side_effect = _query_side_effect

        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
            with pytest.raises(ValueError, match="模板不存在"):
                _validate_template_exists("tmpl_nonexistent")


# ── P1: 一键生成 clip 级效果层映射 ───────────────────────────────────────────