File size: 8,233 Bytes
2d74db3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
"""
Tests for video_processor.py
- extract_keyframes: tested with a real temp directory + mocked subprocess
- encode_image_to_base64: tested with a real temp file
- generate_vertical_reel_clip: filter-chain logic tested without running FFmpeg
- analyze_scene_with_gemini / run_openai_fallback: patched at the boundary
"""
import os
import sys
import glob
import base64
import tempfile
import subprocess
import pytest
from unittest.mock import patch, MagicMock, call

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../app")))

import video_processor
from video_processor import (
    encode_image_to_base64,
    extract_keyframes,
    generate_vertical_reel_clip,
)


# ─── encode_image_to_base64 ───────────────────────────────────────────────────

class TestEncodeImageToBase64:
    def test_encodes_real_file(self, tmp_path):
        img = tmp_path / "frame.jpg"
        img.write_bytes(b"\xff\xd8\xff")   # minimal JPEG header bytes
        result = encode_image_to_base64(str(img))
        # Result must be a valid base64 string that decodes back to our bytes
        assert base64.b64decode(result) == b"\xff\xd8\xff"

    def test_returns_string(self, tmp_path):
        img = tmp_path / "frame.jpg"
        img.write_bytes(b"abc")
        assert isinstance(encode_image_to_base64(str(img)), str)

    def test_missing_file_raises(self):
        with pytest.raises(FileNotFoundError):
            encode_image_to_base64("/nonexistent/path/frame.jpg")


# ─── extract_keyframes ────────────────────────────────────────────────────────

class TestExtractKeyframes:

    @patch("video_processor.subprocess.run")
    def test_creates_output_dir_if_missing(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        output_dir = tmp_path / "frames"
        extract_keyframes("fake.mp4", str(output_dir), interval_seconds=5)
        assert output_dir.exists()

    @patch("video_processor.subprocess.run")
    def test_clears_existing_jpgs_before_extraction(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        output_dir = tmp_path / "frames"
        output_dir.mkdir()
        # Pre-populate with stale files
        (output_dir / "keyframe_0001.jpg").write_bytes(b"old")
        (output_dir / "keyframe_0002.jpg").write_bytes(b"old")
        extract_keyframes("fake.mp4", str(output_dir), interval_seconds=5)
        # After the call (before ffmpeg actually writes) the old files are gone
        assert len(list(output_dir.glob("*.jpg"))) == 0

    @patch("video_processor.subprocess.run")
    def test_ffmpeg_command_uses_correct_fps_filter(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        output_dir = tmp_path / "frames"
        extract_keyframes("fake.mp4", str(output_dir), interval_seconds=10)
        cmd = mock_run.call_args[0][0]
        assert "fps=1/10" in cmd

    @patch("video_processor.subprocess.run")
    def test_ffmpeg_uses_high_quality_jpeg(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        output_dir = tmp_path / "frames"
        extract_keyframes("fake.mp4", str(output_dir), interval_seconds=5)
        cmd = mock_run.call_args[0][0]
        assert "-q:v" in cmd and "2" in cmd

    @patch("video_processor.subprocess.run")
    def test_raises_on_ffmpeg_failure(self, mock_run, tmp_path):
        mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg")
        output_dir = tmp_path / "frames"
        with pytest.raises(subprocess.CalledProcessError):
            extract_keyframes("bad.mp4", str(output_dir))


# ─── generate_vertical_reel_clip ─────────────────────────────────────────────

class TestGenerateVerticalReelClip:

    @patch("video_processor.subprocess.run")
    def test_blurred_mode_uses_boxblur_filter(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")          # must exist β€” validation guard checks os.path.exists
        out = str(tmp_path / "reel.mp4")
        generate_vertical_reel_clip(str(src), 0.0, 30.0, out, render_mode="blurred")
        cmd = mock_run.call_args[0][0]
        vf_value = cmd[cmd.index("-vf") + 1]
        assert "boxblur" in vf_value
        assert "split" in vf_value

    @patch("video_processor.subprocess.run")
    def test_center_mode_uses_center_crop(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        out = str(tmp_path / "reel.mp4")
        generate_vertical_reel_clip(str(src), 0.0, 30.0, out, render_mode="center")
        cmd = mock_run.call_args[0][0]
        vf_value = cmd[cmd.index("-vf") + 1]
        assert "(iw-ow)/2" in vf_value

    @patch("video_processor.subprocess.run")
    def test_left_mode_uses_zero_x_offset(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        out = str(tmp_path / "reel.mp4")
        generate_vertical_reel_clip(str(src), 10.0, 40.0, out, render_mode="left")
        cmd = mock_run.call_args[0][0]
        vf_value = cmd[cmd.index("-vf") + 1]
        assert "crop" in vf_value
        assert ":0," in vf_value or vf_value.endswith(":0")

    @patch("video_processor.subprocess.run")
    def test_right_mode_uses_iw_ow_offset(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        out = str(tmp_path / "reel.mp4")
        generate_vertical_reel_clip(str(src), 5.0, 25.0, out, render_mode="right")
        cmd = mock_run.call_args[0][0]
        vf_value = cmd[cmd.index("-vf") + 1]
        assert "iw-ow" in vf_value

    @patch("video_processor.subprocess.run")
    def test_output_uses_libx264_and_aac(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        out = str(tmp_path / "reel.mp4")
        generate_vertical_reel_clip(str(src), 0.0, 30.0, out)
        cmd = mock_run.call_args[0][0]
        assert "libx264" in cmd
        assert "aac" in cmd

    @patch("video_processor.subprocess.run")
    def test_removes_existing_output_before_render(self, mock_run, tmp_path):
        mock_run.return_value = MagicMock(returncode=0)
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        out = tmp_path / "reel.mp4"
        out.write_bytes(b"old content")
        generate_vertical_reel_clip(str(src), 0.0, 10.0, str(out))
        assert not out.exists()

    @patch("video_processor.subprocess.run")
    def test_raises_on_ffmpeg_failure(self, mock_run, tmp_path):
        mock_run.side_effect = subprocess.CalledProcessError(
            1, "ffmpeg", stderr=b"encoding error"
        )
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        with pytest.raises(subprocess.CalledProcessError):
            generate_vertical_reel_clip(str(src), 0.0, 30.0, str(tmp_path / "out.mp4"))

    def test_raises_when_end_before_start(self, tmp_path):
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        with pytest.raises(ValueError, match="end_time"):
            generate_vertical_reel_clip(str(src), 30.0, 10.0, str(tmp_path / "out.mp4"))

    def test_raises_when_negative_start(self, tmp_path):
        src = tmp_path / "src.mp4"
        src.write_bytes(b"fake")
        with pytest.raises(ValueError, match="negative"):
            generate_vertical_reel_clip(str(src), -5.0, 10.0, str(tmp_path / "out.mp4"))

    def test_raises_when_source_missing(self, tmp_path):
        with pytest.raises(FileNotFoundError):
            generate_vertical_reel_clip("nonexistent.mp4", 0.0, 10.0, str(tmp_path / "out.mp4"))