"""VideoFile 单元测试:probe / iter_frames / frame_times_s / 打不开报错。 旧测试读一段真实样例视频的字节;这里在测试里用 cv2 现造一段几十帧的小视频,按路径读。 旧版先把字节落临时文件,那三条「临时文件有没有删」的测试随之不再适用,已删。 """ from pathlib import Path import cv2 import numpy as np import pytest from cadx.ceus import VideoFile _FRAMES = 40 _WIDTH, _HEIGHT = 64, 48 @pytest.fixture(scope="module") def sample_path(tmp_path_factory: pytest.TempPathFactory) -> Path: """用 cv2 现造一段 40 帧、64×48、25 fps 的 mp4,每帧亮度不同。""" path = tmp_path_factory.mktemp("video") / "sample.mp4" writer = cv2.VideoWriter(str(path), cv2.VideoWriter.fourcc(*"mp4v"), 25.0, (_WIDTH, _HEIGHT)) for index in range(_FRAMES): frame = np.full((_HEIGHT, _WIDTH, 3), (index * 6) % 256, dtype=np.uint8) frame[index % _HEIGHT, :, 2] = 255 writer.write(frame) writer.release() return path def test_probe_returns_positive_metadata(sample_path: Path) -> None: """probe 返回 (fps, frame_count, width, height):字段顺序/类型/语义范围均须正确。""" video = VideoFile(sample_path) result = video.probe() assert isinstance(result, tuple) and len(result) == 4 fps, frame_count, width, height = result # 字段类型契约(app/ports.py 的 DecodedVideo):fps 为 float,其余为 int assert isinstance(fps, float) assert isinstance(frame_count, int) assert isinstance(width, int) assert isinstance(height, int) # 语义合理范围——并钉死字段顺序未颠倒(fps 是小帧率、尺寸是较大像素值) assert 0 < fps < 1000 # 帧率不会是几百几千;若与 frame_count(881) 颠倒会触发 assert frame_count > 0 assert width > 1 and height > 1 def test_probe_field_order_matches_iter(sample_path: Path) -> None: """probe 的 (width, height) 须与 iter_frames 实际帧的 (W, H) 一致(防字段顺序颠倒)。""" video = VideoFile(sample_path) _fps, _frame_count, width, height = video.probe() _idx, frame = next(video.iter_frames()) assert frame.shape == (height, width, 3) def test_iter_frames_indices_contiguous_and_count_matches_probe( sample_path: Path, ) -> None: """iter_frames 产 (idx, frame):idx 从 0 连续递增,且帧数 == probe 的 frame_count。""" video = VideoFile(sample_path) _, frame_count, _, _ = video.probe() indices = [idx for idx, _frame in video.iter_frames()] assert indices == list(range(len(indices))) assert len(indices) == frame_count def test_iter_frames_yields_bgr_uint8_hwc(sample_path: Path) -> None: """每帧为 (H, W, 3) uint8 的 ndarray,且 H/W 与 probe 一致。""" video = VideoFile(sample_path) _, _, width, height = video.probe() idx, frame = next(video.iter_frames()) assert idx == 0 assert isinstance(frame, np.ndarray) assert frame.dtype == np.uint8 assert frame.shape == (height, width, 3) def test_frame_times_are_strictly_increasing_and_match_frame_count( sample_path: Path, ) -> None: video = VideoFile(sample_path) _fps, frame_count, _width, _height = video.probe() times = video.frame_times_s() assert times.shape == (frame_count,) assert times[0] >= 0.0 assert np.all(np.diff(times) > 0.0) def test_解两遍逐帧像素完全相同(sample_path: Path) -> None: """**这条是接缝改成「打开一次」的全部理由**(design/0024)。 CEUS 那条线要分几趟过同一段视频,每趟只留这一趟的小结果。第二趟必须从第 0 帧重新开始, 而且看到的像素要和第一趟一模一样——多趟方案整个建立在这个前提上。它今天成立是因为 每趟各开一个 VideoCapture 顺序读同一个文件,但**这件事此前没有任何一处验证过**。 **比的是全片,而且是流式比。** 只比前几帧的话,可变帧率或者 seek 行为导致的漂移多半 出现在后面而不是开头。两趟各自逐帧过、逐帧喂进哈希,峰值内存是两帧——**这条测试自己 也不许把视频攒起来**,它守的正是这件事。 """ import hashlib def _digest(video: VideoFile) -> tuple[str, int]: h = hashlib.blake2b(digest_size=16) count = 0 for _, frame in video.iter_frames(): h.update(np.ascontiguousarray(frame).tobytes()) count += 1 return h.hexdigest(), count video = VideoFile(sample_path) first = _digest(video) second = _digest(video) assert first[1] > 0 assert first == second, "同一段视频解两遍,逐帧像素必须完全相同" def test_两趟各自从第零帧开始互不干扰(sample_path: Path) -> None: """两趟同时在飞时各读各的游标。 **断言比的是像素,不是 iter_frames 自己报的下标**——那个下标是本模块里一个 Python 计数器,就算内层真把外层的读位置搅乱了,它照样从 0 数上去。 """ video = VideoFile(sample_path) outer = video.iter_frames() _, outer_frame_0 = next(outer) # 内层整个跑起来,中途丢掉——它不该动到外层的游标 inner = video.iter_frames() _, inner_frame_0 = next(inner) _, inner_frame_1 = next(inner) del inner _, outer_frame_1 = next(outer) reference = [frame for _, frame in zip(video.iter_frames(), range(2), strict=False)] assert np.array_equal(inner_frame_0, outer_frame_0), "内层没有从第 0 帧开始" assert np.array_equal(outer_frame_1, inner_frame_1), "外层被内层带跑了:它接着读到的不是第 1 帧" assert len(reference) == 2 def test_bad_bytes_raise_not_silent(tmp_path: Path) -> None: """坏文件无法打开 → 报错(不静默返回空 probe / 空迭代器)。""" garbage = tmp_path / "garbage.mp4" garbage.write_bytes(b"this is not a video file" * 64) video = VideoFile(garbage) with pytest.raises(RuntimeError): video.probe() with pytest.raises(RuntimeError): list(video.iter_frames())