import io import json from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import Mock, patch from model_tooling.video import VideoError, atomic_json, launch, local_path, parser, probe_video, validate_options, safe_stem, storage_base, reserve_run from model_tooling.video_reporting import candidate_counts, class_rows, TerminalProgress, ProgressWriter from model_tooling.video_worker import budget, child, display_box, geometry, read_frame class VideoTests(unittest.TestCase): def args(self, *extra): return parser().parse_args(['--mode', 'fp16', '--video', 'input.mp4', '--output', 'new', *extra]) def test_default_full_and_segment_clamp(self): info = {'duration': 30} self.assertEqual(validate_options(self.args(), info), 30) self.assertEqual(validate_options(self.args('--start', '20', '--duration', '20'), info), 10) def test_bad_numbers(self): for flags in [('--fps', 'nan'), ('--fps', '0'), ('--threshold', '2'), ('--start', '-1'), ('--start', '30'), ('--duration', 'inf'), ('--duration', '0'), ('--timeout', '-1'), ('--panel-width', '241'), ('--max-output-gib', '0')]: with self.subTest(flags=flags), self.assertRaises(VideoError): validate_options(self.args(*flags), {'duration': 30}) def test_paths_and_symlink(self): with tempfile.TemporaryDirectory() as tmp: p = Path(tmp)/'한글 영상.mp4' p.touch() self.assertEqual(local_path(p), p) alias = Path(tmp)/'alias' alias.symlink_to(p) with self.assertRaises(VideoError): local_path(alias) with self.assertRaises(VideoError): local_path(Path(tmp)/'comma,name') def test_existing_output_and_model_mismatch_no_mutation(self): with tempfile.TemporaryDirectory() as tmp: video, output, model = Path(tmp)/'v.mp4', Path(tmp)/'out', Path(tmp)/'m.plan' video.touch(); model.write_bytes(b'not a model') args = self.args('--video', str(video), '--output', str(output), '--fp16-plan', str(model)) with patch('model_tooling.video.probe_video', return_value={'duration': 30}), self.assertRaises(VideoError): launch(args) self.assertFalse(output.exists()) output.mkdir() with patch('model_tooling.video.probe_video', return_value={'duration': 30}), self.assertRaises(VideoError): launch(args) self.assertEqual(list(output.iterdir()), []) def test_invalid_codec_sanitized(self): with patch('model_tooling.video.subprocess.run', side_effect=subprocess.CalledProcessError(1, 'private-path')): with self.assertRaises(VideoError) as ctx: probe_video('input') self.assertNotIn('private-path', str(ctx.exception)) def test_atomic_status(self): with tempfile.TemporaryDirectory() as tmp: target = Path(tmp)/'status.json' atomic_json(target, {'status': 'FAILED'}) self.assertEqual(json.loads(target.read_text())['status'], 'FAILED') self.assertFalse(target.with_suffix('.partial').exists()) def test_read_exact_eof_truncation(self): self.assertEqual(read_frame(io.BytesIO(b'1234'), 4), b'1234') self.assertIsNone(read_frame(io.BytesIO(), 4)) with self.assertRaises(VideoError): read_frame(io.BytesIO(b'12'), 4) def test_frame_memory_short_reads(self): pipe = Mock() pipe.read.side_effect = [b'1', b'2', b'34'] self.assertEqual(read_frame(pipe, 4), b'1234') def test_geometry_and_source_coordinates(self): self.assertEqual(geometry(1280, 720, 960), (960, 540)) self.assertEqual(geometry(640, 480, 960), (640, 480)) self.assertLessEqual(geometry(720, 4000, 960)[1], 1080) self.assertEqual(display_box(128, 72, 640, 360, 1280, 720, 960, 540), (96, 54, 576, 324)) self.assertEqual(display_box(1279.9, 719.9, .1, .1, 1280, 720, 960, 540), (959, 539, 959, 539)) def test_child_cleanup_on_failure_and_interrupt(self): for exception in (VideoError('failure'), KeyboardInterrupt()): proc = Mock(stdin=io.BytesIO(), stdout=io.BytesIO()) proc.poll.return_value = None with patch('model_tooling.video_worker.subprocess.Popen', return_value=proc): with self.assertRaises(type(exception)): with child(['test']): raise exception proc.terminate.assert_called_once() proc.wait.assert_called_once_with(timeout=5) def test_disk_floor_and_output_cap(self): with tempfile.TemporaryDirectory() as tmp, patch('model_tooling.video_worker.OUT', Path(tmp)): with patch('shutil.disk_usage', return_value=Mock(free=0)), self.assertRaises(VideoError): budget({'max_output_bytes': 100}) (Path(tmp)/'out').write_bytes(b'1234') with self.assertRaises(VideoError): budget({'max_output_bytes': 4}) def test_host_interrupt_stops_only_owned_container(self): from model_tooling.video import IMAGE, MODELS with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) video, model, output = root/'v.mp4', root/'m.plan', root/'out' video.touch(); model.touch() (root/'venv/bin').mkdir(parents=True) (root/'venv/bin/python').touch() args = self.args('--video', str(video), '--output', str(output), '--fp16-plan', str(model)) calls = [] def run(command, **kwargs): calls.append(command) if command[1] == 'image': return Mock(stdout=IMAGE+'\n') return Mock(returncode=0) def interrupted(command, *args): calls.append(command) raise KeyboardInterrupt() with patch('model_tooling.video.ENV', root), patch('model_tooling.video.CACHE', root/'cache'), \ patch('model_tooling.video.probe_video', return_value={'duration': 2, 'width': 640, 'height': 360}), \ patch('model_tooling.video.sha256', return_value=MODELS['fp16'][1]), \ patch('model_tooling.video.subprocess.run', side_effect=run), \ patch('model_tooling.video.run_worker', side_effect=interrupted), self.assertRaises(VideoError): launch(args) status = list(output.glob('*/status.json')) self.assertEqual(len(status), 1) self.assertEqual(json.loads(status[0].read_text())['status'], 'INTERRUPTED') started = next(cmd for cmd in calls if cmd[1] == 'run') name = started[started.index('--name')+1] self.assertEqual(calls[-1], ['docker', 'stop', '-t', '5', name]) self.assertTrue(name.startswith('contextra-video-')) self.assertIn('none', started) def test_default_output_and_name_limits(self): with tempfile.TemporaryDirectory() as tmp: video = Path(tmp)/'한글 영상.mp4' base, stem, _ = storage_base(video, None) self.assertEqual(base, video.parent/('컨텍스트라_tensorrt_비교영상_'+stem)) self.assertFalse(base.exists()) long = safe_stem('한'*200) self.assertLessEqual(len(long.encode()), 120) self.assertNotEqual(long, safe_stem('한'*199+'글')) self.assertNotIn('/', safe_stem('../bad:name')) def test_run_collision_preserves_existing(self): with tempfile.TemporaryDirectory() as tmp: base = Path(tmp)/'new'/'base' first = reserve_run(base, 'video', 'compare', '20260910_120000') (first/'keep').write_text('original') second = reserve_run(base, 'video', 'compare', '20260910_120000') self.assertNotEqual(first, second) self.assertTrue(second.name.startswith(first.name+'_')) self.assertEqual((first/'keep').read_text(), 'original') def test_counts_and_class_limit(self): rows = [[0,.5], [0,.49], [1,.8], [2,.1]] result = candidate_counts(rows, .5) self.assertEqual((result['total'],result['shown'],result['hidden']), (4,2,2)) self.assertEqual(result['classes']['0'], {'total':2,'shown':1,'hidden':1}) displayed, omitted = class_rows(result, ['person','bicycle','car'], 2) self.assertEqual(displayed[0], ('person',1,1)) self.assertEqual(omitted, 1) self.assertEqual(candidate_counts([], .5)['total'], 0) def test_tty_and_bounded_non_tty_progress(self): stream = io.StringIO() terminal = TerminalProgress(stream) for n in range(100): terminal.update({'stage':'fp16','current':n,'total':100}, now=n/10) terminal.update({'stage':'fp16','current':100,'total':100,'done':True}, now=10) self.assertEqual(len(stream.getvalue().splitlines()), 2) stream = io.StringIO() stream.isatty = lambda: True terminal = TerminalProgress(stream) terminal.update({'stage':'render','current':1,'total':2}, now=0) terminal.update({'stage':'render','current':2,'total':2,'done':True}, now=1) terminal.finish() self.assertIn('\r\x1b[2K',stream.getvalue()) self.assertEqual(stream.getvalue().count('\n'),1) def test_progress_file_atomic(self): with tempfile.TemporaryDirectory() as tmp: p = Path(tmp)/'progress.json' progress = ProgressWriter(p) progress.update('fp32', 3, 3, done=True) self.assertEqual(json.loads(p.read_text())['current'],3) self.assertFalse(p.with_suffix('.partial').exists()) def test_progress_runner_timeout_reaps_child(self): from model_tooling.video import run_worker with tempfile.TemporaryDirectory() as tmp: process = Mock() process.poll.return_value = None with patch('model_tooling.video.TerminalProgress'), \ patch('model_tooling.video.subprocess.Popen', return_value=process), \ patch('model_tooling.video.time.monotonic', side_effect=[0, 2]), \ self.assertRaises(subprocess.TimeoutExpired): run_worker(['test'], Path(tmp), 1) process.terminate.assert_called_once() process.wait.assert_called_once_with(timeout=5) if __name__ == '__main__': unittest.main()