xiaoxia-saas/tests/unit/test_video_validation.py

"""测试 video_validation 模块 — 渲染输出校验."""

from __future__ import annotations

import json
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
from video_processing.video_validation import (
    FFMPEG_EXIT_CODES,
    VideoValidationResult,
    _check_moov_atom,
    get_exit_code_message,
    validate_video_output,
)

# ── get_exit_code_message 测试 ────────────────────────────────────────────────


class TestGetExitCodeMessage:
    """FFmpeg 退出码映射测试."""

    def test_known_exit_codes(self):
        """已知退出码返回有意义的描述."""
        assert "成功" in get_exit_code_message(0)
        assert "通用错误" in get_exit_code_message(1)
        assert "OOM" in get_exit_code_message(137)
        assert "段错误" in get_exit_code_message(139)
        assert "滤镜错误" in get_exit_code_message(183)
        assert "素材异常" in get_exit_code_message(234)

    def test_signal_termination(self):
        """信号终止（exit > 128 且不在映射表中）返回信号编号."""
        msg = get_exit_code_message(130)  # SIGINT = 130 - 128 = 2
        assert "信号 2" in msg

    def test_unknown_exit_code(self):
        """未知退出码返回通用错误."""
        msg = get_exit_code_message(42)
        assert "未知错误" in msg
        assert "42" in msg

    def test_all_mapped_codes_have_description(self):
        """所有映射的退出码都有名称和描述."""
        for code, (name, desc) in FFMPEG_EXIT_CODES.items():
            assert name, f"exit_code {code} 缺少名称"
            assert desc, f"exit_code {code} 缺少描述"


# ── _check_moov_atom 测试 ─────────────────────────────────────────────────────


class TestCheckMoovAtom:
    """moov atom 检测测试."""

    def test_file_with_moov_in_head(self):
        """文件头部包含 moov 标记 → True（faststart 模式）."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            # 写入一些假数据，中间包含 moov 标记
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 1000)
            f.flush()
            path = Path(f.name)

        try:
            assert _check_moov_atom(path) is True
        finally:
            path.unlink()

    def test_file_with_moov_in_tail(self):
        """文件尾部包含 moov 标记 → True（普通 MP4 模式）."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            # 写一个较大的文件，moov 在尾部
            f.write(b"\x00" * 100_000)
            f.write(b"moov")
            f.write(b"\x00" * 100)
            f.flush()
            path = Path(f.name)

        try:
            assert _check_moov_atom(path) is True
        finally:
            path.unlink()

    def test_file_without_moov(self):
        """文件不含 moov 标记 → False（截断/损坏的 MP4）."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            # 写入纯随机数据，不含 moov
            f.write(b"\x00\x01\x02\x03" * 1000)
            f.flush()
            path = Path(f.name)

        try:
            assert _check_moov_atom(path) is False
        finally:
            path.unlink()

    def test_empty_file(self):
        """空文件 → False."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            path = Path(f.name)

        try:
            assert _check_moov_atom(path) is False
        finally:
            path.unlink()

    def test_tiny_file(self):
        """极小文件（< 8 bytes）→ False."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00\x01")
            f.flush()
            path = Path(f.name)

        try:
            assert _check_moov_atom(path) is False
        finally:
            path.unlink()

    def test_nonexistent_file(self):
        """不存在的文件 → False."""
        assert _check_moov_atom(Path("/nonexistent/file.mp4")) is False


# ── validate_video_output 测试 ────────────────────────────────────────────────


class TestValidateVideoOutput:
    """渲染输出完整性校验测试."""

    def test_file_not_exists(self):
        """文件不存在 → valid=False."""
        result = validate_video_output("/nonexistent/video.mp4")
        assert result.valid is False
        assert result.file_exists is False
        assert "不存在" in result.error_message

    def test_empty_file(self):
        """空文件 → valid=False."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "大小" in result.error_message or "过小" in result.error_message
        finally:
            path.unlink()

    def test_tiny_file(self):
        """极小文件（< 1KB）→ valid=False."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 500)
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "过小" in result.error_message
        finally:
            path.unlink()

    def test_file_without_moov(self):
        """文件有大小但无 moov atom → valid=False."""
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00\x01\x02\x03" * 2000)  # 8KB, 无 moov
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "moov" in result.error_message.lower()
        finally:
            path.unlink()

    @patch("video_processing.video_validation.subprocess.run")
    def test_valid_video(self, mock_run):
        """完整的有效视频 → valid=True."""
        # mock ffprobe 返回
        mock_result = MagicMock()
        mock_result.stdout = json.dumps(
            {
                "streams": [
                    {
                        "width": 1080,
                        "height": 1920,
                        "duration": "10.5",
                        "codec_name": "h264",
                    }
                ],
                "format": {"duration": "10.5"},
            }
        )
        mock_result.returncode = 0
        mock_run.return_value = mock_result

        # 创建含 moov 的文件
        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 5000)
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is True
            assert result.file_exists is True
            assert result.moov_atom_found is True
            assert result.has_video_stream is True
            assert result.width == 1080
            assert result.height == 1920
            assert result.duration == 10.5
            assert result.error_message == ""
        finally:
            path.unlink()

    @patch("video_processing.video_validation.subprocess.run")
    def test_no_video_stream(self, mock_run):
        """文件有 moov 但无视频流 → valid=False."""
        mock_result = MagicMock()
        mock_result.stdout = json.dumps({"streams": [], "format": {"duration": "0"}})
        mock_result.returncode = 0
        mock_run.return_value = mock_result

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 5000)
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "无视频流" in result.error_message
        finally:
            path.unlink()

    @patch("video_processing.video_validation.subprocess.run")
    def test_ffprobe_timeout(self, mock_run):
        """ffprobe 超时 → valid=False."""
        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffprobe", timeout=15)

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 5000)
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "超时" in result.error_message
        finally:
            path.unlink()

    @patch("video_processing.video_validation.subprocess.run")
    def test_ffprobe_error(self, mock_run):
        """ffprobe 执行失败 → valid=False."""
        mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd="ffprobe", stderr="Invalid data found")

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 5000)
            path = Path(f.name)

        try:
            result = validate_video_output(path)
            assert result.valid is False
            assert "ffprobe" in result.error_message.lower()
        finally:
            path.unlink()

    @patch("video_processing.video_validation.subprocess.run")
    def test_short_video_warning(self, mock_run):
        """极短视频（< min_duration）→ 仍 valid=True 但有警告."""
        mock_result = MagicMock()
        mock_result.stdout = json.dumps(
            {
                "streams": [{"width": 100, "height": 100, "duration": "0.05", "codec_name": "h264"}],
                "format": {"duration": "0.05"},
            }
        )
        mock_result.returncode = 0
        mock_run.return_value = mock_result

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            f.write(b"\x00" * 100)
            f.write(b"moov")
            f.write(b"\x00" * 5000)
            path = Path(f.name)

        try:
            # 默认 min_duration=0.1，视频只有 0.05s → 应该仍然 valid（仅警告）
            result = validate_video_output(path)
            assert result.valid is True
        finally:
            path.unlink()


# ── VideoValidationResult 数据结构测试 ────────────────────────────────────────


class TestVideoValidationResult:
    """VideoValidationResult 数据结构测试."""

    def test_default_invalid(self):
        """默认构造结果为无效."""
        result = VideoValidationResult(valid=False)
        assert result.valid is False
        assert result.is_valid is False
        assert result.file_size == 0
        assert result.error_message == ""

    def test_valid_result(self):
        """有效结果属性正确."""
        result = VideoValidationResult(
            valid=True,
            file_exists=True,
            file_size=1024000,
            moov_atom_found=True,
            has_video_stream=True,
            duration=15.3,
            width=1920,
            height=1080,
        )
        assert result.is_valid is True
        assert result.duration == 15.3
        assert result.width == 1920