from __future__ import annotations import csv from datetime import datetime, timedelta, timezone import hashlib import importlib.util import io import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import textwrap import unittest from unittest import mock import zlib import numpy as np from PIL import Image, ImageDraw MODULE_PATH = Path(__file__).resolve().parents[1] / "extract_ppt_slides.py" SPEC = importlib.util.spec_from_file_location("extract_ppt_slides", MODULE_PATH) assert SPEC and SPEC.loader slides = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = slides SPEC.loader.exec_module(slides) def palette(marker_bin: int | None = None, coverage: float = 0.20): histogram = np.full(slides.PALETTE_BIN_COUNT, 1 / slides.PALETTE_BIN_COUNT, np.float32) if marker_bin is not None: histogram.fill((1.0 - coverage) / (slides.PALETTE_BIN_COUNT - 1)) histogram[marker_bin] = coverage histogram /= float(np.sum(histogram, dtype=np.float64)) return histogram def feature( index: int, array: np.ndarray, score: float = 0.2, sharpness: float | None = None, page_palette: np.ndarray | None = None, ): value = int(np.mean(array)) phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big") return slides.FrameFeature( sample_index=index, relative_seconds=float(index), gray=array.astype(np.uint8).copy(), palette_histogram=(palette() if page_palette is None else page_palette.copy()), perceptual_hash=phash, content_score=score, sharpness=float(value if sharpness is None else sharpness), ) def candidate( timestamp: float, array: np.ndarray, score: float, phash: int = 0, page_palette: np.ndarray | None = None, ): return slides.PageCandidate( timestamp, array.astype(np.uint8).copy(), palette() if page_palette is None else page_palette.copy(), phash, score, 1.0, ) def logical_page(order: int, page_palette: np.ndarray | None = None): page_feature = np.full((8, 8), order % 256, np.uint8) return slides.LogicalPage( logical_anchor_feature=page_feature.copy(), current_feature=page_feature, current_palette=palette() if page_palette is None else page_palette.copy(), current_hash=order, current_timestamp=float(order), current_score=0.2, first_seen_order=order, ) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest().upper() def pdf_rgb_pages(path: Path): data = path.read_bytes() objects = { int(match.group(1)): match.group(2) for match in re.finditer(rb"(?m)^(\d+) 0 obj\n(.*?)\nendobj\n", data, re.DOTALL) } pages_body = objects[2] kids_match = re.search(rb"/Kids \[(.*?)\]", pages_body, re.DOTALL) assert kids_match page_ids = [int(value) for value in re.findall(rb"(\d+) 0 R", kids_match.group(1))] result = [] for page_id in page_ids: page = objects[page_id] media = re.search(rb"/MediaBox \[0 0 (\d+) (\d+)\]", page) image_ref = re.search(rb"/Im0 (\d+) 0 R", page) assert media and image_ref image_body = objects[int(image_ref.group(1))] length = int(re.search(rb"/Length (\d+)", image_body).group(1)) marker = b"stream\n" start = image_body.index(marker) + len(marker) compressed = image_body[start : start + length] result.append(((int(media.group(1)), int(media.group(2))), zlib.decompress(compressed))) return result class ShortReadStream: def __init__(self, chunks): self.chunks = list(chunks) def read(self, size): if not self.chunks: return b"" chunk = self.chunks.pop(0) if len(chunk) <= size: return chunk self.chunks.insert(0, chunk[size:]) return chunk[:size] class ReadExactTests(unittest.TestCase): def test_short_reads_clean_eof_and_partial_eof(self): self.assertEqual(slides.read_exact(ShortReadStream([b"a", b"bc", b"def"]), 6), b"abcdef") self.assertIsNone(slides.read_exact(ShortReadStream([]), 6)) with self.assertRaisesRegex(slides.SlideExtractionError, "半帧 EOF"): slides.read_exact(ShortReadStream([b"abc"]), 6) class SegmentAndClassifierTests(unittest.TestCase): def test_anchor_drift_emits_only_old_and_new_plateaus(self): detector = slides.StableSegmentDetector() emitted = [] arrays = [np.full((8, 8), 20, np.uint8)] * 3 arrays += [np.full((8, 8), value, np.uint8) for value in (22, 24, 26, 28, 30, 32)] arrays += [np.full((8, 8), 80, np.uint8)] * 3 for index, array in enumerate(arrays): value = detector.add(feature(index, array, score=index / 100, sharpness=index)) if value: emitted.append(value) final = detector.flush() if final: emitted.append(final) self.assertEqual([round(float(np.mean(item.feature))) for item in emitted], [20, 80]) def test_animation_increase_decrease_and_equal_score(self): classifier = slides.PageClassifier() base = np.zeros((8, 8), np.uint8) added = base.copy() added[:, :2] = 20 self.assertEqual(classifier.consume(candidate(1, base, 0.20)), "new_page") self.assertEqual(classifier.consume(candidate(2, added, 0.30)), "animation_updated") self.assertEqual(classifier.pages[0].current_timestamp, 2) reduced = base.copy() self.assertEqual(classifier.consume(candidate(3, reduced, 0.10)), "animation_retained") self.assertEqual(classifier.pages[0].current_timestamp, 2) self.assertEqual( classifier.consume(candidate(4, added, 0.30 + slides.CONTENT_SCORE_TOLERANCE / 2)), "animation_updated", ) self.assertEqual(classifier.pages[0].current_timestamp, 4) def test_a_b_a_keeps_first_order(self): classifier = slides.PageClassifier() a = np.zeros((8, 8), np.uint8) b = np.full((8, 8), 255, np.uint8) classifier.consume(candidate(1, a, 0.2, 0)) classifier.consume(candidate(2, b, 0.2, (1 << 64) - 1)) action = classifier.consume(candidate(3, a, 0.2, 0)) self.assertEqual(action, "historical_duplicate") self.assertEqual(len(classifier.pages), 2) self.assertEqual([page.first_seen_order for page in classifier.pages], [0, 1]) def test_recent_animation_has_priority_over_historical_duplicate(self): classifier = slides.PageClassifier() historical = np.zeros((8, 8), np.uint8) recent = np.full((8, 8), 4, np.uint8) classifier.consume(candidate(1, historical, 0.20, 0)) classifier.pages.append( slides.LogicalPage(recent.copy(), recent.copy(), palette(), 0, 2.0, 0.20, 1) ) action = classifier.consume(candidate(3, historical, 0.22, 0)) self.assertEqual(action, "animation_updated") self.assertEqual(len(classifier.pages), 2) class MainSequenceTests(unittest.TestCase): def test_palette_histogram_distinguishes_dark_blue_from_black(self): image = np.array( [ [[16, 16, 48], [16, 16, 48]], [[16, 16, 16], [16, 16, 16]], ], dtype=np.uint8, ) histogram = slides._palette_histogram(image) self.assertEqual(histogram.shape, (512,)) self.assertTrue(np.all(np.isfinite(histogram))) self.assertAlmostEqual(float(np.sum(histogram)), 1.0, places=6) self.assertEqual(float(histogram[1]), 0.5) self.assertEqual(float(histogram[0]), 0.5) def test_selects_old_pages_002_through_038_and_keeps_two_gaps(self): marker_pages = {2, *range(4, 9), *range(10, 39)} pages = [ logical_page(index - 1, palette(1) if index in marker_pages else palette()) for index in range(1, 54) ] selected = slides.select_main_sequence(pages) self.assertEqual(selected, pages[1:38]) self.assertIs(selected[1], pages[2]) # old slide_003: first one-page gap self.assertIs(selected[7], pages[8]) # old slide_009: second one-page gap self.assertEqual([page.first_seen_order for page in selected], list(range(1, 38))) def test_conservative_gates_keep_all_candidates(self): cases = { "markers": [palette(1) if index < 7 else palette() for index in range(10)], "share": [palette(1) if 3 <= index <= 10 else palette() for index in range(20)], "density": [ palette(1) if index in {0, 2, 4, 6, 7, 8, 9, 10} else palette() for index in range(11) ], } for name, histograms in cases.items(): with self.subTest(name=name): pages = [logical_page(index, value) for index, value in enumerate(histograms)] selected = slides.select_main_sequence(pages) self.assertEqual(selected, pages) self.assertTrue(all(left is right for left, right in zip(selected, pages))) def test_all_theme_pages_are_preserved(self): pages = [logical_page(index, palette(7)) for index in range(12)] selected = slides.select_main_sequence(pages) self.assertEqual(selected, pages) self.assertTrue(all(left is right for left, right in zip(selected, pages))) def test_invalid_palette_is_explicit_failure(self): invalid = [ np.zeros(5, np.float32), np.full(512, np.nan, np.float32), np.zeros(512, np.float32), ] for value in invalid: with self.subTest(shape=value.shape, finite=bool(np.all(np.isfinite(value)))): with self.assertRaises(slides.SlideExtractionError): slides.select_main_sequence([logical_page(0, value)]) class ProcessAndCommandTests(unittest.TestCase): def test_command_freezes_time_mapping(self): scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180) self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan) input_index = scan.index("-i") self.assertLess(scan.index("-threads"), input_index) self.assertEqual(scan[scan.index("-threads") + 1], str(slides.MEDIA_DECODE_THREADS)) self.assertEqual(scan[scan.index("-filter_threads") + 1], str(slides.MEDIA_FILTER_THREADS)) extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png")) self.assertEqual(extract[extract.index("-ss") + 1], "12.000") self.assertLess(extract.index("-threads"), extract.index("-i")) self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0) def test_hardware_commands_place_decoder_and_threads_before_input(self): for codec, decoder in (("h264", "h264_cuvid"), ("hevc", "hevc_cuvid")): with self.subTest(codec=codec): self.assertEqual(slides._decoder_for_codec(codec, "yuv420p"), decoder) command = slides._scan_command( "ffmpeg", Path(f"{codec}.mp4"), 320, 180, decoder ) input_index = command.index("-i") for option in ("-hwaccel", "-hwaccel_output_format", "-c:v", "-threads"): self.assertLess(command.index(option), input_index) self.assertEqual(command[command.index("-hwaccel") + 1], "cuda") self.assertEqual(command[command.index("-c:v") + 1], decoder) self.assertEqual( command[command.index("-threads") + 1], str(slides.MEDIA_DECODE_THREADS), ) self.assertIn("hwdownload,format=nv12", command[command.index("-vf") + 1]) self.assertEqual(slides._decoder_for_codec("h264", "gbrp"), "h264_cuvid") self.assertEqual(slides._decoder_for_codec("hevc", "p010le"), "hevc_cuvid") self.assertIsNone(slides._decoder_for_codec("vp9", "yuv420p")) def test_hardware_external_failures_have_typed_boundary(self): class FakeStdout: def __init__(self, payloads): self.payloads = list(payloads) def read(self, size): return self.payloads.pop(0) if self.payloads else b"" def close(self): return None class FakeProcess: def __init__(self, payloads, code): self.stdout = FakeStdout(payloads) self.code = code def wait(self, timeout=None): return self.code def poll(self): return self.code def terminate(self): return None def kill(self): return None info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") cases = ( ("nonzero", FakeProcess([], 7), "exit=7"), ("zero", FakeProcess([], 0), "未输出任何完整"), ("partial", FakeProcess([b"x", b""], 0), "输出半帧"), ) with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_which_or_error", return_value="ffmpeg" ): root = Path(raw) with mock.patch.object(slides.subprocess, "Popen", side_effect=OSError("start denied")): with self.assertRaisesRegex(slides.HardwareDecodePathError, "启动失败"): slides._scan_video_once(Path("x.mp4"), root, info, decoder="h264_cuvid") for name, process, pattern in cases: with self.subTest(name=name), mock.patch.object( slides.subprocess, "Popen", return_value=process ): with self.assertRaisesRegex(slides.HardwareDecodePathError, pattern): slides._scan_video_once( Path("x.mp4"), root, info, decoder="h264_cuvid" ) def test_hardware_cleanup_failure_does_not_trigger_cpu_fallback(self): info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") hardware_error = slides.HardwareDecodePathError("hardware-root") slides._attach_cleanup_note(hardware_error, "cleanup-failed") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", side_effect=hardware_error ) as scan: with self.assertRaises(slides.HardwareDecodePathError) as caught: slides.scan_video(Path("x.mp4"), Path(raw), info) self.assertIs(caught.exception, hardware_error) self.assertEqual(scan.call_count, 1) self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__)) def test_active_errors_keep_identity_when_process_cleanup_fails(self): frame_bytes = 320 * 180 * 3 class Stream: def __init__(self, error=None, partial=False): self.error = error self.partial = partial self.calls = 0 def read(self, size): self.calls += 1 if self.error is not None: raise self.error if self.partial: return b"x" if self.calls == 1 else b"" return b"\0" * frame_bytes if self.calls == 1 else b"" def close(self): return None class FakeProcess: def __init__(self, stream): self.stdout = stream def wait(self, timeout=None): return 0 def poll(self): return None def terminate(self): return None cleanup = OSError("cleanup-failed") info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") cases = ( ("business", None, ValueError("business-root")), ("cpu", None, slides.SlideExtractionError("cpu-root")), ("base", KeyboardInterrupt(), None), ) for name, read_error, feature_error in cases: root_error = read_error or feature_error stream = Stream(error=read_error) patches = [ mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess(stream)), mock.patch.object(slides, "_attempt_reap", return_value=cleanup), ] if feature_error is not None: patches.append(mock.patch.object(slides, "_gray_feature", side_effect=feature_error)) with self.subTest(name=name), tempfile.TemporaryDirectory() as raw: for item in patches: item.start() try: with self.assertRaises(type(root_error)) as caught: slides._scan_video_once( Path("x.mp4"), Path(raw), info, decoder=None ) self.assertIs(caught.exception, root_error) self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__)) finally: for item in reversed(patches): item.stop() partial_process = FakeProcess(Stream(partial=True)) with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_which_or_error", return_value="ffmpeg" ), mock.patch.object(slides.subprocess, "Popen", return_value=partial_process), mock.patch.object( slides, "_attempt_reap", return_value=cleanup ): with self.assertRaises(slides.HardwareDecodePathError) as caught: slides._scan_video_once( Path("x.mp4"), Path(raw), info, decoder="h264_cuvid" ) hardware_error = caught.exception self.assertTrue(getattr(hardware_error, "_media_cleanup_failed", False)) self.assertIn("cleanup-failed", "\n".join(hardware_error.__notes__)) with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", side_effect=hardware_error ) as scan: with self.assertRaises(slides.HardwareDecodePathError) as caught_again: slides.scan_video(Path("x.mp4"), Path(raw), info) self.assertIs(caught_again.exception, hardware_error) self.assertEqual(scan.call_count, 1) def test_product_popen_uses_below_normal_creation_flags(self): class FakeProcess: stdout = None def wait(self, timeout=None): return 0 def poll(self): return 0 with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_subprocess_creation_flags", return_value=12345 ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()) as popen: slides._run_file_command(["tool"], Path(raw) / "stderr.log", "stage") self.assertEqual(popen.call_args.kwargs["creationflags"], 12345) def test_fallback_only_handles_hardware_error_and_never_retries_cpu(self): info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") page = logical_page(0) success = slides.ScanResult((page,), 10, 2, (("new_page", 1),), None) hardware_error = slides.HardwareDecodePathError("decoder failed") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", side_effect=[hardware_error, success] ) as scan, mock.patch("sys.stderr", new_callable=io.StringIO) as stderr: self.assertEqual(slides.scan_video(Path("x.mp4"), Path(raw), info), [page]) self.assertEqual([call.kwargs["decoder"] for call in scan.call_args_list], [ "h264_cuvid", None ]) self.assertIn("decoder failed", stderr.getvalue()) cpu_error = slides.SlideExtractionError("cpu root cause") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", side_effect=[hardware_error, cpu_error] ) as scan: with self.assertRaises(slides.SlideExtractionError) as caught: slides.scan_video(Path("x.mp4"), Path(raw), info) self.assertIs(caught.exception, cpu_error) self.assertEqual(scan.call_count, 2) def test_algorithm_empty_and_baseexception_never_fall_back(self): info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") empty = slides.ScanResult((), 10, 0, (), "h264_cuvid") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", return_value=empty ) as scan: with self.assertRaisesRegex(slides.SlideExtractionError, "未发现"): slides.scan_video(Path("x.mp4"), Path(raw), info) scan.assert_called_once() for error in (RuntimeError("classifier failed"), KeyboardInterrupt(), SystemExit(9)): with self.subTest(error=type(error).__name__), tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_scan_video_once", side_effect=error ) as scan: with self.assertRaises(type(error)) as caught: slides.scan_video(Path("x.mp4"), Path(raw), info) self.assertIs(caught.exception, error) scan.assert_called_once() def test_algorithm_exception_inside_hardware_scan_preserves_original_type(self): frame_bytes = 320 * 180 * 3 class FullFrameStream: def __init__(self): self.done = False def read(self, size): if self.done: return b"" self.done = True return b"\0" * frame_bytes def close(self): return None class FakeProcess: stdout = FullFrameStream() def wait(self, timeout=None): return 0 def poll(self): return 0 root_error = ValueError("feature root cause") info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides, "_which_or_error", return_value="ffmpeg" ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()), mock.patch.object( slides, "_gray_feature", side_effect=root_error ): with self.assertRaises(ValueError) as caught: slides._scan_video_once( Path("x.mp4"), Path(raw), info, decoder="h264_cuvid" ) self.assertIs(caught.exception, root_error) @unittest.skipUnless(shutil.which("powershell"), "Windows PowerShell is required") def test_resource_sampler_parser_rejects_cpu_boundaries_and_empty_windows(self): sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1" fields = ( "timestamp_utc", "previous_valid_timestamp_utc", "interval_seconds", "counter_warmup", "cpu_valid", "host_cpu_percent", "scan_phase_active", "scan_ffmpeg_present", "scan_ffmpeg_exit_observed", "scan_ffmpeg_pid", "scan_ffmpeg_command_line", "decoder", "logical_processor_count", "task_cpu_percent", "task_working_set_bytes", "task_pids", "root_priority", "ffmpeg_priorities", "ffmpeg_command_lines", "gpu_query_status", "gpu_query_raw_rows", "gpu_utilization_percent", "gpu_decoder_utilization_percent", "gpu_memory_used_mib", "nvidia_smi_compute_query_status", "nvidia_smi_compute_raw_rows", "nvidia_smi_compute_task_rows", "gpu_process_memory_query_status", "gpu_process_memory_instances", "task_gpu_dedicated_memory_mib", ) scan_command = ( "ffmpeg -hwaccel cuda -hwaccel_output_format cuda " "-c:v h264_cuvid -threads 4 " "-filter_threads 2 -filter_complex_threads 2 -i x.mp4 -map 0:v:0 " "-vf hwdownload,format=nv12,fps=fps=1:start_time=0 " "-pix_fmt rgb24 -f rawvideo -" ) def make_row(current, previous, interval, cpu, *, warmup, present, exited): return { "timestamp_utc": current.isoformat().replace("+00:00", "Z"), "previous_valid_timestamp_utc": ( "" if previous is None else previous.isoformat().replace("+00:00", "Z") ), "interval_seconds": f"{interval:.6f}", "counter_warmup": str(warmup).lower(), "cpu_valid": str(not warmup).lower(), "host_cpu_percent": "" if warmup else f"{cpu:.6f}", "scan_phase_active": "true", "scan_ffmpeg_present": str(present).lower(), "scan_ffmpeg_exit_observed": str(exited).lower(), "scan_ffmpeg_pid": "4242", "scan_ffmpeg_command_line": scan_command, "decoder": "h264_cuvid", "logical_processor_count": "44", "task_cpu_percent": "" if warmup else "12.5", "task_working_set_bytes": "123456789", "task_pids": "101;4242", "root_priority": "BelowNormal", "ffmpeg_priorities": ( "[]" if not present else '[{"pid":4242,"priority":"BelowNormal"}]' ), "ffmpeg_command_lines": ( '[{"pid":4242,"command_line":' + json.dumps(scan_command) + "}]" ), "gpu_query_status": "ok", "gpu_query_raw_rows": '["20, 31, 2048"]', "gpu_utilization_percent": "20", "gpu_decoder_utilization_percent": "31", "gpu_memory_used_mib": "2048", "nvidia_smi_compute_query_status": "ok", "nvidia_smi_compute_raw_rows": '["4242, N/A"]', "nvidia_smi_compute_task_rows": '["4242, N/A"]', "gpu_process_memory_query_status": "ok", "gpu_process_memory_instances": ( '[{"pid":4242,"instance_name":"pid_4242_luid_0",' '"counter_path":"path","dedicated_usage_bytes":1048576}]' ), "task_gpu_dedicated_memory_mib": "1", } def analyze_case(root, name, intervals, cpu_values, mutate=None, omit_field=None): case = root / name case.mkdir() samples = case / "samples.csv" current = datetime(2026, 8, 4, tzinfo=timezone.utc) rows = [make_row(current, None, 0, 0, warmup=True, present=True, exited=False)] previous = current for interval, cpu in zip(intervals, cpu_values): current += timedelta(seconds=interval) rows.append(make_row( current, previous, interval, cpu, warmup=False, present=True, exited=False, )) previous = current current += timedelta(seconds=0.5) rows.append(make_row( current, previous, 0.5, 20.0, warmup=False, present=False, exited=True, )) if mutate is not None: mutate(rows) output_fields = tuple(field for field in fields if field != omit_field) with samples.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=output_fields, extrasaction="ignore") writer.writeheader() writer.writerows(rows) evidence = case / "evidence" result = subprocess.run( [ shutil.which("powershell"), "-NoProfile", "-File", str(sampler), "-AnalyzeSamplesPath", str(samples), "-EvidenceDirectory", str(evidence), ], capture_output=True, text=True, timeout=60, ) self.assertEqual(result.returncode, 0, result.stderr + result.stdout) return json.loads((evidence / "resource_summary.json").read_text(encoding="utf-8-sig")) with tempfile.TemporaryDirectory() as raw: root = Path(raw) low = analyze_case(root, "low", [0.5] * 22, [84.9] * 22) self.assertTrue(low["sampling_valid"]) self.assertGreater(low["window_count"], 0) self.assertTrue(low["cpu_gate_pass"]) equal = analyze_case(root, "equal", [0.5] * 22, [85.0] * 22) self.assertTrue(equal["sampling_valid"]) self.assertFalse(equal["all_windows_host_cpu_strictly_below_85"]) self.assertFalse(equal["cpu_gate_pass"]) exact_ten = analyze_case( root, "exact-ten", [0.5] * 21, [85.0] * 20 + [20.0] ) self.assertEqual(exact_ten["maximum_high_cpu_segment_seconds"], 10.0) self.assertFalse(exact_ten["cpu_gate_pass"]) below_ten = analyze_case( root, "below-ten", [0.49995] * 20 + [0.5], [85.0] * 20 + [20.0] ) self.assertEqual(below_ten["maximum_high_cpu_segment_seconds"], 9.999) sparse = analyze_case(root, "sparse", [1.0] * 12, [20.0] * 12) self.assertFalse(sparse["sampling_valid"]) self.assertTrue(any("window_density_invalid" in item for item in sparse["sampling_problems"])) zero_window = analyze_case(root, "zero-window", [0.5] * 8, [20.0] * 8) self.assertFalse(zero_window["sampling_valid"]) self.assertEqual(zero_window["window_count"], 0) gap = analyze_case( root, "gap", [0.5] * 10 + [1.6] + [0.5] * 10, [20.0] * 21 ) self.assertFalse(gap["sampling_valid"]) self.assertGreater(gap["maximum_observed_interval_seconds"], 1.5) reversed_time = analyze_case( root, "time-reversal", [0.5] * 12, [20.0] * 12, mutate=lambda rows: rows[5].__setitem__("timestamp_utc", rows[3]["timestamp_utc"]), ) self.assertFalse(reversed_time["sampling_valid"]) self.assertTrue(any( "timestamp_not_strictly_increasing" in item for item in reversed_time["sampling_problems"] )) interval_mismatch = analyze_case( root, "interval-mismatch", [0.5] * 12, [20.0] * 12, mutate=lambda rows: rows[4].__setitem__("interval_seconds", "9.0"), ) self.assertFalse(interval_mismatch["sampling_valid"]) self.assertTrue(any( "interval_mismatch" in item for item in interval_mismatch["sampling_problems"] )) previous_mismatch = analyze_case( root, "previous-mismatch", [0.5] * 12, [20.0] * 12, mutate=lambda rows: rows[4].__setitem__( "previous_valid_timestamp_utc", rows[1]["timestamp_utc"] ), ) self.assertFalse(previous_mismatch["sampling_valid"]) self.assertTrue(any( "previous_timestamp_mismatch" in item for item in previous_mismatch["sampling_problems"] )) for omitted in ( "gpu_query_raw_rows", "task_pids", "ffmpeg_command_lines", "nvidia_smi_compute_task_rows", ): with self.subTest(omitted=omitted): missing_field = analyze_case( root, f"missing-{omitted}", [0.5] * 12, [20.0] * 12, omit_field=omitted, ) self.assertFalse(missing_field["sampling_valid"]) self.assertIn( f"required_column_missing_{omitted}", missing_field["sampling_problems"], ) first_not_present = analyze_case( root, "first-coverage", [0.5] * 12, [20.0] * 12, mutate=lambda rows: rows[0].__setitem__("scan_ffmpeg_present", "false"), ) self.assertFalse(first_not_present["sampling_valid"]) self.assertIn("scan_phase_first_row_not_present", first_not_present["sampling_problems"]) last_not_exit = analyze_case( root, "last-coverage", [0.5] * 12, [20.0] * 12, mutate=lambda rows: rows[-1].__setitem__("scan_ffmpeg_exit_observed", "false"), ) self.assertFalse(last_not_exit["sampling_valid"]) self.assertIn("scan_phase_last_row_not_exit_observation", last_not_exit["sampling_problems"]) argv_rewrites = { "missing-output-format": lambda value: value.replace( "-hwaccel_output_format cuda ", "" ), "wrong-output-format": lambda value: value.replace( "-hwaccel_output_format cuda", "-hwaccel_output_format qsv" ), "late-output-format": lambda value: value.replace( "-hwaccel_output_format cuda ", "" ).replace("-i x.mp4", "-i x.mp4 -hwaccel_output_format cuda"), "missing-hwdownload": lambda value: value.replace( "hwdownload,format=nv12,", "" ), } for name, rewrite in argv_rewrites.items(): with self.subTest(argv_case=name): invalid_argv = analyze_case( root, f"invalid-{name}", [0.5] * 12, [20.0] * 12, mutate=lambda rows, transform=rewrite: [ row.__setitem__( "scan_ffmpeg_command_line", transform(row["scan_ffmpeg_command_line"]), ) for row in rows ], ) self.assertFalse(invalid_argv["sampling_valid"]) self.assertIn("nvdec_scan_argv_invalid", invalid_argv["sampling_problems"]) @unittest.skipUnless( os.name == "nt" and shutil.which("powershell") and shutil.which("nvidia-smi"), "Windows PowerShell and nvidia-smi are required", ) def test_resource_sampler_run_rejects_forged_nvdec_and_excludes_page_ffmpeg(self): sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1" sampler_text = sampler.read_text(encoding="utf-8") process_start = sampler_text.index("[void]$Process.Start()") self.assertLess( sampler_text.index("$HostCpuCounter = New-Object"), process_start ) self.assertLess(sampler_text.index("$HostCpuCounter.NextValue()"), process_start) self.assertLess( sampler_text.index("$GpuMemoryCategory = New-Object"), process_start ) self.assertLess(sampler_text.index("$GpuMemoryCategory.GetInstanceNames()"), process_start) self.assertLess( sampler_text.index('$StartInfo.EnvironmentVariables["MBX_MEDIA_SAMPLER_READY"]'), process_start, ) with tempfile.TemporaryDirectory() as raw: root = Path(raw) fake_ffmpeg = root / "ffmpeg.exe" shutil.copy2(sys.executable, fake_ffmpeg) child = root / "fake_ffmpeg_child.py" child.write_text( "import sys, time\n" "time.sleep(float(sys.argv[sys.argv.index('--sleep') + 1]))\n", encoding="utf-8", ) target = root / "stub_media_target.py" target.write_text( textwrap.dedent( f""" import ctypes import os from pathlib import Path import subprocess import sys import time sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00004000) creationflags = 0x00004000 if os.environ.get("MBX_MEDIA_SAMPLER_READY") != "1": raise RuntimeError("sampler-ready handshake missing") fake_ffmpeg = Path({str(fake_ffmpeg)!r}) child = Path({str(child)!r}) scan = subprocess.Popen([ str(fake_ffmpeg), str(child), "--sleep", "7.0", "-hwaccel", "cuda", "-c:v", "h264_cuvid", "-threads", "4", "-filter_threads", "2", "-filter_complex_threads", "2", "-i", "fake.mp4", "-map", "0:v:0", "-vf", "hwdownload,format=nv12,setpts=PTS-STARTPTS," "fps=fps=1:start_time=0,scale=320:180", "-pix_fmt", "rgb24", "-f", "rawvideo", "-", ], creationflags=creationflags) scan.wait() print("警告：NVDEC 不可用；改用受限 CPU。", file=sys.stderr, flush=True) print( "扫描解码路径：NVDEC/CUDA (h264_cuvid)；解码线程=4；滤镜线程=2", flush=True, ) time.sleep(1.5) for index in range(24): churn = subprocess.Popen([ str(fake_ffmpeg), str(child), "--sleep", "0.04", "-threads", "4", "-i", "fake.mp4", "-frames:v", "1", "-n", f"race-{{index}}.png", ], creationflags=creationflags) churn.wait() page = subprocess.Popen([ str(fake_ffmpeg), str(child), "--sleep", "4.0", "-threads", "4", "-i", "fake.mp4", "-frames:v", "1", "-n", "fake.png", ], creationflags=creationflags) page.wait() """ ).lstrip(), encoding="utf-8", ) source = root / "source.stub" source.write_bytes(b"read-only source fingerprint") evidence = root / "evidence" result = subprocess.run( [ shutil.which("powershell"), "-NoProfile", "-File", str(sampler), "-PythonPath", sys.executable, "-ScriptPath", str(target), "-VideoPath", str(source), "-OutputPath", str(root / "unused-output"), "-EvidenceDirectory", str(evidence), ], capture_output=True, text=True, encoding="utf-8", errors="replace", env={ **os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1", "MBX_MEDIA_SAMPLER_TEST_CIM_DELAY_MS": "100", "MBX_MEDIA_SAMPLER_TEST_FAIL_SAMPLE_ONCE": "1", }, timeout=90, ) self.assertEqual(result.returncode, 3, result.stderr + result.stdout) run_summary = json.loads( (evidence / "run_summary.json").read_text(encoding="utf-8-sig") ) self.assertEqual(run_summary["target_exit_code"], 0) self.assertFalse(run_summary["resource_acceptance_pass"]) self.assertTrue( run_summary["nvdec_command_log_evidence"], json.dumps( { "run_summary": run_summary, "stdout": (evidence / "stdout.log").read_text("utf-8"), }, ensure_ascii=False, ), ) self.assertFalse(run_summary["nvdec_scan_argv_evidence"]) self.assertGreater(run_summary["command_line_disappearance_race_count"], 0) self.assertEqual(run_summary["injected_sample_failure_count"], 1) self.assertEqual( len(run_summary["sampling_errors"]), 1, run_summary["sampling_errors"] ) self.assertIn("injected post-scan sample failure", run_summary["sampling_errors"][0]) self.assertTrue(run_summary["source_unchanged"]) self.assertIn("NVDEC/CUDA (h264_cuvid)", (evidence / "stdout.log").read_text("utf-8")) self.assertIn("改用受限 CPU", (evidence / "stderr.log").read_text("utf-8")) with (evidence / "resource_samples.csv").open( newline="", encoding="utf-8-sig" ) as handle: rows = list(csv.DictReader(handle)) resource_summary = json.loads( (evidence / "resource_summary.json").read_text(encoding="utf-8-sig") ) self.assertFalse(resource_summary["sampling_valid"], resource_summary) self.assertEqual( resource_summary["sampling_problems"], ["nvdec_scan_argv_invalid"] ) self.assertTrue(resource_summary["all_windows_host_cpu_strictly_below_85"]) self.assertTrue(resource_summary["window_count"] > 0) self.assertFalse(any( "previous_timestamp" in problem or "interval_mismatch" in problem for problem in resource_summary["sampling_problems"] )) for index, row in enumerate(rows): if index == 0: self.assertEqual(row["previous_valid_timestamp_utc"], "") self.assertAlmostEqual(float(row["interval_seconds"]), 0.0, places=6) continue current = datetime.fromisoformat(row["timestamp_utc"].replace("Z", "+00:00")) previous = datetime.fromisoformat( rows[index - 1]["timestamp_utc"].replace("Z", "+00:00") ) reported_previous = datetime.fromisoformat( row["previous_valid_timestamp_utc"].replace("Z", "+00:00") ) self.assertLess(abs((reported_previous - previous).total_seconds()), 0.001) self.assertLess( abs(float(row["interval_seconds"]) - (current - previous).total_seconds()), 0.001, ) active_indexes = [ index for index, row in enumerate(rows) if row["scan_phase_active"].lower() == "true" ] self.assertGreaterEqual( len(active_indexes), 2, json.dumps({"rows": rows, "run_summary": run_summary}, ensure_ascii=False), ) self.assertEqual(len({rows[index]["scan_ffmpeg_pid"] for index in active_indexes}), 1) last_active = active_indexes[-1] self.assertEqual(rows[last_active]["scan_ffmpeg_present"].lower(), "false") self.assertEqual(rows[last_active]["scan_ffmpeg_exit_observed"].lower(), "true") later_rows = rows[last_active + 1 :] self.assertTrue(later_rows) self.assertTrue(any("-frames:v" in row["ffmpeg_command_lines"] for row in later_rows)) self.assertTrue(all(row["scan_phase_active"].lower() == "false" for row in later_rows)) @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required") def test_h264_h265_nvdec_and_cpu_product_chains_are_equivalent(self): ffmpeg = shutil.which("ffmpeg") capabilities = subprocess.check_output( [ffmpeg, "-hide_banner", "-encoders"], text=True, errors="replace" ) + subprocess.check_output( [ffmpeg, "-hide_banner", "-decoders"], text=True, errors="replace" ) required = ("libx264", "libx265", "h264_cuvid", "hevc_cuvid") missing = [name for name in required if name not in capabilities] if missing: self.skipTest(f"missing FFmpeg capabilities: {missing}") with tempfile.TemporaryDirectory() as raw: root = Path(raw) frames_dir = root / "frames" frames_dir.mkdir() size = (320, 180) def page(kind, animated=False): colors = {"a": (245, 245, 240), "b": (20, 45, 85), "c": (232, 210, 70)} image = Image.new("RGB", size, colors[kind]) draw = ImageDraw.Draw(image) draw.rectangle((0, 0, 319, 30), fill=(25, 90, 160)) draw.rectangle((30, 55, 285, 70), fill=(30, 30, 30)) if kind != "a" or animated: draw.rectangle((30, 95, 250, 110), fill=(35, 35, 35)) if animated: draw.rectangle((260, 140, 280, 160), fill=(210, 30, 30)) return image transition_colors = ((100, 100, 100), (120, 25, 130), (20, 150, 90), (180, 65, 20)) frames = [page("a")] * 3 + [Image.new("RGB", size, transition_colors[0])] frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[1])] frames += [page("b")] * 3 + [Image.new("RGB", size, transition_colors[2])] frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[3])] frames += [page("c")] * 3 for index, image in enumerate(frames): image.save(frames_dir / f"frame_{index:03d}.png") codec_settings = { "h264": ("libx264", "h264_cuvid"), "hevc": ("libx265", "hevc_cuvid"), } evidence = [] for codec, (encoder, decoder) in codec_settings.items(): source = root / f"sequence-{codec}.mp4" command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-framerate", "1", "-i", str(frames_dir / "frame_%03d.png"), "-threads", "4", "-c:v", encoder, "-preset", "ultrafast", ] if codec == "h264": command.extend(("-qp", "0")) else: command.extend(("-x265-params", "lossless=1:pools=4")) command.extend(("-pix_fmt", "yuv420p", "-n", str(source))) subprocess.run(command, check=True, timeout=120) probe_staging = root / f"probe-{codec}" probe_staging.mkdir() info = slides.probe_video(source, probe_staging) self.assertEqual(info.codec_name, codec) self.assertEqual(info.pixel_format, "yuv420p") hardware_staging = root / f"scan-{codec}-hardware" cpu_staging = root / f"scan-{codec}-cpu" hardware_staging.mkdir() cpu_staging.mkdir() hardware = slides._scan_video_once( source, hardware_staging, info, decoder=decoder ) cpu = slides._scan_video_once(source, cpu_staging, info, decoder=None) self.assertEqual(hardware.sample_count, cpu.sample_count) self.assertEqual(hardware.stable_candidate_count, cpu.stable_candidate_count) self.assertEqual(hardware.classification_counts, cpu.classification_counts) self.assertEqual(len(hardware.pages), len(cpu.pages)) hardware_selected = slides.select_main_sequence(hardware.pages) cpu_selected = slides.select_main_sequence(cpu.pages) self.assertEqual( [page.first_seen_order for page in hardware_selected], [page.first_seen_order for page in cpu_selected], ) self.assertEqual(len(hardware_selected), len(cpu_selected)) self.assertGreaterEqual(len(hardware_selected), 3) for hardware_page, cpu_page in zip(hardware_selected, cpu_selected): self.assertAlmostEqual( hardware_page.current_timestamp, cpu_page.current_timestamp, delta=0.001, ) product_summaries = [] for label, selected in (("hardware", hardware_selected), ("cpu", cpu_selected)): output = root / f"product-{codec}-{label}" output.mkdir() pngs = slides.extract_original_pages(source, output, info, selected) pdf = output / "slides.pdf" slides.write_image_pdf(pngs, pdf) png_summary = [] for path in pngs: with Image.open(path) as image: png_summary.append((path.name, image.mode, image.size)) product_summaries.append((png_summary, pdf_rgb_pages(pdf))) self.assertEqual(product_summaries[0][0], product_summaries[1][0]) self.assertEqual( [page[0] for page in product_summaries[0][1]], [page[0] for page in product_summaries[1][1]], ) self.assertEqual( len(product_summaries[0][1]), len(product_summaries[1][1]) ) hardware_command = slides._scan_command( ffmpeg, source, *slides._scan_size(info), decoder ) cpu_command = slides._scan_command( ffmpeg, source, *slides._scan_size(info), None ) evidence.extend(( { "codec": codec, "path": "nvdec", "decoder": decoder, "argv": hardware_command, "samples": hardware.sample_count, "stable_candidates": hardware.stable_candidate_count, "classifications": dict(hardware.classification_counts), "selected_pages": len(hardware_selected), "timestamps": [page.current_timestamp for page in hardware_selected], }, { "codec": codec, "path": "cpu", "decoder": None, "argv": cpu_command, "samples": cpu.sample_count, "stable_candidates": cpu.stable_candidate_count, "classifications": dict(cpu.classification_counts), "selected_pages": len(cpu_selected), "timestamps": [page.current_timestamp for page in cpu_selected], }, )) print("MEDIA_HOST_SYNTHETIC_EVIDENCE=" + json.dumps(evidence, ensure_ascii=False)) @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required") def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) video = root / "vfr-nonzero.mkv" subprocess.run( [ shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=red:size=64x64:rate=2:duration=2", "-f", "lavfi", "-i", "color=green:size=64x64:rate=3:duration=2", "-f", "lavfi", "-i", "color=blue:size=64x64:rate=4:duration=2", "-filter_complex", "[0:v][1:v][2:v]concat=n=3:v=1:a=0,setpts=PTS+5/TB[v]", "-map", "[v]", "-fps_mode", "vfr", "-c:v", "ffv1", str(video), ], check=True, ) start_time = float( subprocess.check_output( [ shutil.which("ffprobe"), "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=start_time", "-of", "default=nw=1:nk=1", str(video), ], text=True, ).strip() ) self.assertGreaterEqual(start_time, 5.0) scan_process = subprocess.Popen( slides._scan_command(shutil.which("ffmpeg"), video, 64, 64), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) assert scan_process.stdout is not None scanned = [] while True: payload = slides.read_exact(scan_process.stdout, 64 * 64 * 3) if payload is None: break scanned.append(np.frombuffer(payload, np.uint8).reshape((64, 64, 3))) scan_process.stdout.close() stderr = scan_process.stderr.read() if scan_process.stderr else b"" if scan_process.stderr: scan_process.stderr.close() self.assertEqual(scan_process.wait(), 0, stderr.decode(errors="replace")) self.assertGreaterEqual(len(scanned), 5) extracted = [] for timestamp in (0.0, 2.0, 4.0): path = root / f"at-{timestamp:.0f}.png" slides._run_file_command( slides._extract_command(shutil.which("ffmpeg"), video, timestamp, path), root / f"at-{timestamp:.0f}.stderr.log", "VFR 定位", ) with Image.open(path) as image: extracted.append(np.asarray(image.convert("RGB"), np.uint8)) for sequence in ((scanned[0], scanned[2], scanned[4]), tuple(extracted)): means = [np.mean(item, axis=(0, 1)) for item in sequence] self.assertEqual([int(np.argmax(value)) for value in means], [0, 1, 2]) @unittest.skipUnless(shutil.which("ffmpeg"), "FFmpeg is required") def test_large_stderr_does_not_deadlock_and_nonzero_is_reported(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) ok_log = root / "ok.log" slides._run_file_command( [sys.executable, "-c", "import sys; sys.stderr.write('x'*300000)"], ok_log, "桩命令", ) self.assertEqual(ok_log.stat().st_size, 300000) bad_log = root / "bad.log" with self.assertRaisesRegex(slides.SlideExtractionError, "exit=7"): slides._run_file_command( [sys.executable, "-c", "import sys; sys.stderr.write('boom'); raise SystemExit(7)"], bad_log, "桩命令", ) def test_keyboard_interrupt_reaps_file_process_and_preserves_type(self): events = [] handles = [] class FakeProcess: stdout = None def __init__(self): self.wait_calls = 0 self.running = True def wait(self, timeout=None): self.wait_calls += 1 events.append(("wait", timeout)) if self.wait_calls == 1: raise KeyboardInterrupt() self.running = False return 0 def poll(self): return None if self.running else 0 def terminate(self): events.append(("terminate", None)) def kill(self): events.append(("kill", None)) def make_process(*args, **kwargs): handles.append(kwargs["stderr"]) return FakeProcess() with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") output = root / "result" page = logical_page(0) with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object( slides, "scan_video", return_value=[page] ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object( slides.subprocess, "Popen", side_effect=make_process ): with self.assertRaises(KeyboardInterrupt): slides.extract_slides(str(source), str(output)) self.assertFalse(output.exists()) self.assertFalse(any(root.glob(".result.staging-*"))) self.assertTrue(handles[0].closed) self.assertEqual(events[1][0], "terminate") self.assertEqual(events[2][0], "wait") def test_keyboard_interrupt_reaps_scan_process_and_preserves_type(self): events = [] class Stdout: def read(self, size): raise KeyboardInterrupt() def close(self): events.append("close") class FakeProcess: def __init__(self): self.stdout = Stdout() self.running = True def wait(self, timeout=None): events.append("wait") self.running = False return 0 def poll(self): return None if self.running else 0 def terminate(self): events.append("terminate") def kill(self): events.append("kill") with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") output = root / "result" with mock.patch.object( slides.subprocess, "Popen", return_value=FakeProcess() ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object( slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10) ): with self.assertRaises(KeyboardInterrupt): slides.extract_slides(str(source), str(output)) self.assertFalse(output.exists()) self.assertFalse(any(root.glob(".result.staging-*"))) self.assertEqual(events[:3], ["close", "terminate", "wait"]) def test_stderr_open_failure_is_wrapped_with_stage(self): with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides.Path, "open", side_effect=OSError("denied") ): with self.assertRaisesRegex(slides.SlideExtractionError, "桩阶段失败：denied"): slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段") def test_cleanup_error_does_not_replace_keyboard_interrupt(self): class FakeProcess: stdout = None def wait(self, timeout=None): raise KeyboardInterrupt() def poll(self): return None def terminate(self): raise OSError("cleanup denied") with tempfile.TemporaryDirectory() as raw, mock.patch.object( slides.subprocess, "Popen", return_value=FakeProcess() ): with self.assertRaises(KeyboardInterrupt): slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段") def test_stderr_close_error_preserves_interrupt_in_both_helpers(self): class CloseFailingHandle: def close(self): raise OSError("stderr close denied") def flush(self): return None class FileProcess: stdout = None def __init__(self): self.calls = 0 def wait(self, timeout=None): self.calls += 1 if self.calls == 1: raise KeyboardInterrupt() return 0 def poll(self): return None def terminate(self): return None class ScanStdout: def read(self, size): raise KeyboardInterrupt() def close(self): return None class ScanProcess: def __init__(self): self.stdout = ScanStdout() def wait(self, timeout=None): return 0 def poll(self): return None def terminate(self): return None with tempfile.TemporaryDirectory() as raw: root = Path(raw) with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object( slides.subprocess, "Popen", return_value=FileProcess() ): with self.assertRaises(KeyboardInterrupt): slides._run_file_command(["fake"], root / "file.stderr.log", "文件阶段") with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object( slides.subprocess, "Popen", return_value=ScanProcess() ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"): with self.assertRaises(KeyboardInterrupt): slides.scan_video(Path("x.mp4"), root, slides.VideoInfo(640, 360, 10)) def test_stderr_close_error_is_staged_on_normal_and_ordinary_failure(self): class CloseFailingHandle: def close(self): raise OSError("stderr close denied") def flush(self): return None class FakeProcess: stdout = None def __init__(self, code): self.code = code def wait(self, timeout=None): return self.code def poll(self): return self.code with tempfile.TemporaryDirectory() as raw: root = Path(raw) with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object( slides.subprocess, "Popen", return_value=FakeProcess(0) ): with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段关闭 stderr 失败"): slides._run_file_command(["fake"], root / "normal.log", "文件阶段") with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object( slides.subprocess, "Popen", return_value=FakeProcess(7) ), mock.patch.object(slides, "_tail", return_value="boom"): with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段失败（exit=7）"): slides._run_file_command(["fake"], root / "failed.log", "文件阶段") class PdfTests(unittest.TestCase): def test_pdf_rgb_content_and_order(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) paths = [] for index, color in enumerate(((255, 20, 20), (20, 40, 255))): path = root / f"{index}.png" Image.new("RGB", (17 + index, 11 + index), color).save(path) paths.append(path) pdf = root / "slides.pdf" slides.write_image_pdf(paths, pdf) decoded = pdf_rgb_pages(pdf) self.assertEqual([item[0] for item in decoded], [(17, 11), (18, 12)]) for path, (_, rgb) in zip(paths, decoded): with Image.open(path) as image: self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest()) pdfinfo = shutil.which("pdfinfo") if pdfinfo and Path(pdfinfo).suffix.lower() not in {".cmd", ".bat"}: result = subprocess.run([pdfinfo, str(pdf)], capture_output=True, text=True, check=True) self.assertIn("Pages: 2", result.stdout) def test_twenty_pages_keep_one_decoded_image_active(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) paths = [] for index in range(20): path = root / f"{index:02d}.png" Image.new("RGB", (12, 8), (index, index * 2, index * 3)).save(path) paths.append(path) active = 0 maximum = 0 class CountedOpen: def __init__(self, path): self.image = Image.open(path) def __enter__(self): nonlocal active, maximum active += 1 maximum = max(maximum, active) return self.image def __exit__(self, exc_type, exc, traceback): nonlocal active self.image.close() active -= 1 pdf = root / "twenty.pdf" slides.write_image_pdf(paths, pdf, image_opener=CountedOpen) self.assertEqual(maximum, 1) self.assertEqual(active, 0) self.assertEqual(len(pdf_rgb_pages(pdf)), 20) class SafetyAndIntegrationTests(unittest.TestCase): def test_existing_output_is_rejected_before_external_process(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"not a video") for kind in ("directory", "file"): with self.subTest(kind=kind): output = root / f"existing-{kind}" if kind == "directory": output.mkdir() else: output.write_bytes(b"occupied") with mock.patch.object(slides.subprocess, "Popen") as popen: with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"): slides.extract_slides(str(source), str(output)) popen.assert_not_called() def test_live_and_broken_output_links_are_rejected_by_lexical_precheck(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") for link_kind in ("live", "broken"): with self.subTest(link_kind=link_kind): output = slides._absolute_lexical(root / f"{link_kind}-link") original_lexists = slides._lexists def controlled_lexists(path): return True if path == output else original_lexists(path) with mock.patch.object(slides, "_lexists", side_effect=controlled_lexists), mock.patch.object( slides.subprocess, "Popen" ) as popen: with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"): slides.extract_slides(str(source), str(output)) popen.assert_not_called() def test_input_symlink_is_rejected_before_external_process(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = slides._absolute_lexical(root / "source-link.mp4") with mock.patch.object(slides, "_is_symlink", side_effect=lambda path: path == source), mock.patch.object( slides.subprocess, "Popen" ) as popen: with self.assertRaisesRegex(slides.SlideExtractionError, "拒绝符号链接"): slides.extract_slides(str(source), str(root / "result")) popen.assert_not_called() def test_baseexception_removes_staging_and_leaves_no_formal_output(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") output = root / "result" with mock.patch.object(slides, "probe_video", side_effect=KeyboardInterrupt()): with self.assertRaises(KeyboardInterrupt): slides.extract_slides(str(source), str(output)) self.assertFalse(output.exists()) self.assertFalse(any(root.glob(".result.staging-*"))) def test_main_sequence_filter_runs_once_before_original_extraction(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") output = root / "result" candidates = [logical_page(index, palette(1)) for index in range(10)] selected = candidates[1:9] with mock.patch.object( slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10) ), mock.patch.object( slides, "scan_video", return_value=candidates ), mock.patch.object( slides, "select_main_sequence", return_value=selected ) as sequence_filter, mock.patch.object( slides, "extract_original_pages", return_value=[] ) as original_extraction, mock.patch.object(slides, "write_image_pdf"): self.assertEqual(slides.extract_slides(str(source), str(output)), output) sequence_filter.assert_called_once_with(candidates) self.assertIs(original_extraction.call_args.args[3], selected) def test_selected_pages_are_renumbered_continuously(self): with tempfile.TemporaryDirectory() as raw: staging = Path(raw) selected = [logical_page(index, palette(1)) for index in range(3, 6)] def write_png(command, stderr_path, stage): destination = Path(command[-1]) Image.new("RGB", (64, 36), (20, 40, 80)).save(destination) with mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object( slides, "_run_file_command", side_effect=write_png ): outputs = slides.extract_original_pages( Path("source.mp4"), staging, slides.VideoInfo(64, 36, 10), selected ) self.assertEqual([path.name for path in outputs], [ "slide_001.png", "slide_002.png", "slide_003.png" ]) def test_commit_conflict_is_wrapped_and_staging_is_removed(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) source = root / "source.mp4" source.write_bytes(b"stub") output = root / "result" with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object( slides, "scan_video", return_value=[logical_page(0)] ), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object( slides, "write_image_pdf" ), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")): with self.assertRaisesRegex(slides.SlideExtractionError, "提交正式输出目录失败：race"): slides.extract_slides(str(source), str(output)) self.assertFalse(output.exists()) self.assertFalse(any(root.glob(".result.staging-*"))) @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required") def test_chinese_path_synthetic_video_end_to_end(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) / "中文路径" root.mkdir() frame_root = root / "frames" frame_root.mkdir() size = (640, 360) def page_a(complete: bool): image = Image.new("RGB", size, "white") draw = ImageDraw.Draw(image) draw.rectangle((0, 0, 639, 65), fill=(26, 78, 145)) draw.rectangle((55, 100, 585, 125), fill=(30, 30, 30)) if complete: draw.rectangle((55, 165, 470, 190), fill=(30, 30, 30)) draw.rectangle((55, 230, 540, 255), fill=(30, 30, 30)) draw.rectangle((570, 320, 580, 330), fill=(220, 20, 20)) return image def page_b(): image = Image.new("RGB", size, (18, 34, 68)) draw = ImageDraw.Draw(image) draw.rectangle((65, 75, 575, 285), fill=(230, 190, 45)) draw.rectangle((110, 125, 530, 155), fill=(18, 34, 68)) return image frames = [page_a(False)] * 3 frames += [Image.new("RGB", size, (127, 127, 127))] frames += [page_a(True)] * 3 frames += [Image.new("RGB", size, (90, 20, 130))] frames += [page_b()] * 3 frames += [Image.new("RGB", size, (20, 150, 90))] frames += [page_a(True)] * 3 for index, image in enumerate(frames): image.save(frame_root / f"frame_{index:03d}.png") source = root / "合成会议录屏.mp4" subprocess.run( [ shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error", "-framerate", "1", "-i", str(frame_root / "frame_%03d.png"), "-c:v", "libx264rgb", "-crf", "0", "-pix_fmt", "rgb24", str(source), ], check=True, ) before = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source)) output = root / "验收输出" result = subprocess.run( [sys.executable, str(MODULE_PATH), str(source), "--output", str(output)], capture_output=True, text=True, timeout=120, ) self.assertEqual(result.returncode, 0, result.stderr + result.stdout) self.assertEqual( {entry.name for entry in output.iterdir()}, {"pages", "合成会议录屏.slides.pdf"}, ) pngs = sorted((output / "pages").glob("slide_*.png")) self.assertEqual(len(pngs), 2, result.stdout) for png in pngs: with Image.open(png) as image: self.assertEqual(image.size, size) expected = [page_a(True), page_b()] for actual_path, expected_image in zip(pngs, expected): with Image.open(actual_path) as actual: self.assertEqual( hashlib.sha256(actual.convert("RGB").tobytes()).digest(), hashlib.sha256(expected_image.tobytes()).digest(), ) decoded = pdf_rgb_pages(output / "合成会议录屏.slides.pdf") self.assertEqual(len(decoded), 2) for png, (_, rgb) in zip(pngs, decoded): with Image.open(png) as image: self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest()) after = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source)) self.assertEqual(after, before) self.assertFalse(any(root.glob(".验收输出.staging-*"))) if __name__ == "__main__": unittest.main()