from pathlib import Path import shutil import subprocess import sys import tempfile import unittest import os import json import signal from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools")) import video_clip as vc import video_clip_resources as resources class VideoClipUnitTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.source = self.root / "한글 영상 폴더 #1" / "싸움 테스트 (최종) #1.mp4" self.source.parent.mkdir() self.source.write_bytes(b"source") self.info = vc.VideoInfo( path=self.source.resolve(), codec="h264", width=1920, height=1080, fps=29.97, duration=60.0, has_audio=True, ) def request(self, **kwargs): values = dict(source=self.source, start=10.0, duration=5.0, mode='auto') values.update(kwargs) return vc.ClipRequest(**values) def test_auto_prefers_stream_copy_for_matching_h264(self): mode, reasons = vc.choose_mode(self.info, self.request()) self.assertEqual(mode, "copy") self.assertIn("재인코딩이 필요하지 않음", reasons[0]) def test_auto_reencodes_for_codec_change(self): info = vc.VideoInfo(**{**self.info.__dict__, "codec": "hevc"}) mode, reasons = vc.choose_mode(info, self.request()) self.assertEqual(mode, "reencode") self.assertTrue(any("codec" in reason for reason in reasons)) def test_auto_reencodes_for_resolution_or_fps(self): for req in [ self.request(resolution=(1280, 720)), self.request(fps=10.0), ]: with self.subTest(req=req): mode, reasons = vc.choose_mode(self.info, req) self.assertEqual(mode, "reencode") self.assertTrue(reasons) def test_copy_mode_rejects_transform(self): with self.assertRaisesRegex(vc.VideoClipError, "STREAM COPY"): vc.choose_mode( self.info, self.request(mode="copy", resolution=(1280, 720)), ) def test_letterbox_filter_preserves_aspect_ratio(self): flt, notes = vc._build_filter( self.info, self.request(resolution=(1280, 720)), "reencode", ) self.assertIn("force_original_aspect_ratio=decrease", flt) self.assertIn("pad=1280:720", flt) self.assertIn("setsar=1", flt) self.assertEqual(notes, []) def test_odd_dimensions_get_minimal_padding_on_reencode(self): info = vc.VideoInfo( **{**self.info.__dict__, "width": 1279, "height": 719} ) self.assertEqual( vc._expected_dimensions(info, self.request(), "reencode"), (1280, 720), ) flt, notes = vc._build_filter(info, self.request(), "reencode") self.assertIn("pad=1280:720", flt) self.assertTrue(notes) def test_default_output_keeps_unicode_and_special_filename(self): output = vc._default_output_path(self.info.path, self.request()) self.assertEqual(output.parent, self.source.parent.resolve() / 'clips') self.assertIn("싸움 테스트 (최종) #1_s", output.name) self.assertEqual(output.suffix, ".mp4") def test_ffmpeg_command_uses_argument_list_for_special_path(self): request = self.request(source=self.info.path) command = vc._build_copy_command( self.info, request, self.root / "출력 결과 #1.mp4", ) self.assertIsInstance(command, list) self.assertIn(str(self.info.path), command) self.assertNotIn(" ".join(command), command) def test_run_never_uses_shell(self): fake = subprocess.CompletedProcess( ["ffprobe"], 0, stdout="{}", stderr="", ) with patch.object(vc, "CURRENT_JOB") as current: run = current.get.return_value.run run.return_value = fake vc._run(["ffprobe", "경로 with space.mp4"]) self.assertEqual(run.call_args.kwargs['timeout'], 15) self.assertEqual(run.call_args.args[0][-1], "경로 with space.mp4") def test_resolution_requires_even_dimensions(self): with self.assertRaisesRegex(vc.VideoClipError, "짝수"): vc._normalize_request( self.request(resolution=(1279, 720)), self.info, ) def test_requested_range_must_fit_source(self): with self.assertRaisesRegex(vc.VideoClipError, "초과"): vc._normalize_request( self.request(start=58, duration=5), self.info, ) def test_existing_output_is_not_overwritten_by_default(self): out = self.root / "existing.mp4" out.write_bytes(b"keep") with self.assertRaisesRegex(vc.VideoClipError, "이미 있습니다"): vc._normalize_request( self.request(output=out), self.info, ) self.assertEqual(out.read_bytes(), b"keep") def test_reencode_command_maps_cpu_quality_to_crf(self): request = self.request( source=self.info.path, mode="reencode", resolution=(1280, 720), fps=10.0, quality=21, preset="fast", ) with patch.object(vc, "_encoder_available", return_value=True): command, _ = vc._build_reencode_command( self.info, request, self.root / "out.mp4", ) self.assertIn("libx264", command) self.assertEqual(command[command.index("-crf") + 1], "21") self.assertIn("fps=10", command[command.index("-vf") + 1]) def test_reencode_command_maps_nvenc_quality_to_cq(self): request = self.request( source=self.info.path, mode="reencode", encoder="nvenc", preset="p4", quality=24, ) with patch.object(vc, "_encoder_available", return_value=True): command, _ = vc._build_reencode_command( self.info, request, self.root / "out.mp4", ) self.assertIn("h264_nvenc", command) self.assertEqual(command[command.index("-cq") + 1], "24") def test_output_validation_rejects_audio(self): actual = vc.VideoInfo( path=self.root / "out.mp4", codec="h264", width=1920, height=1080, fps=29.97, duration=5.0, has_audio=True, ) with patch.object(vc, "probe_video", return_value=actual): with self.assertRaisesRegex(vc.VideoClipError, "audio"): vc.validate_output( self.info, self.request(), "copy", actual.path, ) def test_auto_copy_failure_falls_back_to_reencode(self): output = self.root / "결과 파일 #1.mp4" request = self.request(output=output) copy_actual = vc.VideoInfo( path=output, codec="h264", width=1920, height=1080, fps=29.97, duration=7.0, has_audio=False, ) encoded_actual = vc.VideoInfo( path=output, codec="h264", width=1920, height=1080, fps=29.97, duration=5.0, has_audio=False, ) calls = {"ffmpeg": 0} def fake_probe(path): if Path(path).resolve() == self.source.resolve(): return self.info return copy_actual if calls["ffmpeg"] == 1 else encoded_actual def fake_run(args, **kwargs): del kwargs if args[0] == "ffmpeg" and "-encoders" not in args: calls["ffmpeg"] += 1 Path(args[-1]).write_bytes(b"result") return subprocess.CompletedProcess( args, 0, stdout=" V..... libx264 encoder\n", stderr="", ) with ( patch.object(vc.shutil, "which", return_value="/usr/bin/fake"), patch.object(vc, "probe_video", side_effect=fake_probe), patch.object(vc, "_run", side_effect=fake_run), patch.object(vc, "_encoder_available", return_value=True), ): result = vc.clip_video(request) self.assertEqual(result.mode, "reencode") self.assertEqual(calls["ffmpeg"], 2) self.assertTrue(output.exists()) @unittest.skipUnless( shutil.which("ffmpeg") and shutil.which("ffprobe"), "ffmpeg/ffprobe required", ) class VideoClipFfmpegIntegrationTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp = tempfile.TemporaryDirectory() cls.addClassCleanup(cls.temp.cleanup) cls.root = Path(cls.temp.name) / '한글 폴더 #1' cls.root.mkdir() cls.source = cls.root / '원본 영상 (테스트).mp4' with vc.JobGuard() as job: job.output_dir = cls.root if not vc._encoder_available('libx264'): raise unittest.SkipTest('libx264 unavailable') vc._run(['ffmpeg', '-hide_banner', '-loglevel', 'error', '-nostdin', '-n', '-filter_threads', '1', '-f', 'lavfi', '-i', 'testsrc=size=320x240:rate=10:duration=4', '-c:v', 'libx264', '-threads', '2', '-pix_fmt', 'yuv420p', '-bf', '0', '-force_key_frames', 'expr:gte(t,n_forced*1)', '-an', str(cls.source)], timeout=30) def test_real_ffmpeg_unicode_path_auto_stream_copy(self): output = self.root / '출력 영상 #copy.mp4' result = vc.clip_video(vc.ClipRequest(self.source, 1, 1, mode='copy', output=output)) self.assertEqual(result.mode, 'copy') self.assertEqual((result.width, result.height), (320, 240)) self.assertTrue(output.exists()) def test_real_ffmpeg_unicode_path_reencode_and_probe(self): result = vc.clip_video(vc.ClipRequest(self.source, .5, 1, resolution=(640, 360), fps=10)) self.assertEqual(result.mode, 'reencode') self.assertEqual((result.width, result.height), (640, 360)) self.assertAlmostEqual(result.fps or 0, 10.0, delta=.05) self.assertEqual(result.output.parent, self.root / 'clips') self.assertTrue(result.output.exists()) class VideoClipSafetyTest(unittest.TestCase): setUp = VideoClipUnitTest.setUp request = VideoClipUnitTest.request def test_defaults_and_directory_conflict(self): args = vc.build_parser().parse_args(['in.mp4', '--start', '1', '--duration', '2']) self.assertEqual(args.mode, 'reencode') r = vc.ClipRequest(self.source, 1, 2, output_dir=self.root/'out') self.assertEqual(vc._normalize_request(r, self.info).output.parent, self.root/'out') with self.assertRaises(vc.VideoClipError): vc._normalize_request(self.request(output=self.root/'x.mp4', output_dir=self.root/'out'), self.info) def test_symlink_and_source_alias_rejected(self): link = self.root/'link.mp4';link.symlink_to(self.source) with self.assertRaises(vc.VideoClipError): vc._normalize_request(self.request(output=link, overwrite=True),self.info) link.unlink();os.link(self.source,link) with self.assertRaises(vc.VideoClipError): vc._normalize_request(self.request(output=link, overwrite=True),self.info) def test_atomic_no_clobber_and_explicit_replace(self): out=self.root/'out.mp4';temp=self.root/'temp.mp4';temp.write_bytes(b'new') out.write_bytes(b'concurrent') fd=os.open(self.root,os.O_RDONLY|os.O_DIRECTORY) try: with self.assertRaises(vc.VideoClipError):vc._publish(temp,out,False,fd) self.assertEqual(out.read_bytes(),b'concurrent') vc._publish(temp,out,True,fd) self.assertEqual(out.read_bytes(),b'new') finally:os.close(fd) def test_operational_failure_does_not_fallback(self): for error in [vc.CommandError('permission','Permission denied'),vc.CommandError('disk','No space left on device; could not find tag for codec'),vc.VideoClipError('timeout'),KeyboardInterrupt()]: with self.subTest(error=type(error).__name__), \ patch.object(vc,'probe_video',return_value=self.info), \ patch.object(vc,'_run',side_effect=error) as run: out=self.root/'old.mp4';out.write_bytes(b'keep') with self.assertRaises(type(error)): vc.clip_video(self.request(output=out,overwrite=True)) self.assertEqual(run.call_count,1) self.assertEqual(out.read_bytes(),b'keep') self.assertFalse(list(self.root.glob('.contextra-clip-*'))) def test_cgroup_bound_membership_and_lock(self): with vc.JobGuard() as job: group=job.cgroup self.assertEqual((group/'cpu.max').read_text().strip(),'200000 100000') self.assertEqual((group/'memory.max').read_text().strip(),str(2*1024**3)) with self.assertRaisesRegex(vc.VideoClipError,'다른 영상 작업'): with vc.JobGuard():pass result=job.run([sys.executable,'-c','print(open("/proc/self/cgroup").read())']) self.assertIn(group.name,result.stdout) self.assertFalse(group.exists()) def test_timeout_and_descendant_cleanup(self): with vc.JobGuard() as job: group=job.cgroup with self.assertRaisesRegex(vc.VideoClipError,'시간 상한'): job.run([sys.executable,'-c','import subprocess,time; subprocess.Popen(["sleep","30"]);time.sleep(30)'],timeout=.15) self.assertEqual((group/'cgroup.procs').read_text().strip(),'') self.assertFalse(group.exists()) def test_signal_cancel_and_log_limit(self): with vc.JobGuard() as job: with self.assertRaisesRegex(vc.VideoClipError,'취소'): job.run([sys.executable,'-c','import os,signal,time;os.kill(os.getppid(),signal.SIGTERM);time.sleep(30)']) self.assertEqual((job.cgroup/'cgroup.procs').read_text().strip(),'') with self.assertRaisesRegex(vc.VideoClipError,'로그'): job.run([sys.executable,'-c','import sys;sys.stdout.write("x"*2000000)']) def test_disk_guards(self): job=vc.JobGuard();job.output_dir=self.root usage=shutil.disk_usage(self.root) with patch.object(resources.shutil,'disk_usage',return_value=usage._replace(free=1)): with self.assertRaisesRegex(vc.VideoClipError,'15GiB'):job.check_disk() with patch.object(resources.shutil,'disk_usage',side_effect=[usage._replace(free=resources.ROOT_FLOOR+1),usage._replace(free=1)]): with self.assertRaisesRegex(vc.VideoClipError,'2GiB'):job.check_disk() def test_unavailable_resources_fail_closed(self): with patch.object(resources.os,'geteuid',return_value=1000): with self.assertRaises(vc.VideoClipError): with vc.JobGuard():pass def test_controller_setup_failure_does_not_spawn(self): with patch.object(resources.Path,'mkdir',side_effect=PermissionError('controller unavailable')), \ patch.object(resources.subprocess,'Popen') as spawn: with self.assertRaises(vc.VideoClipError): with vc.JobGuard():pass spawn.assert_not_called() if __name__ == "__main__": unittest.main()