tests/test_segments.py
Rama master · réplica de lectura del repositorio original.
"""HLS 分片：关键帧对齐、时间戳连续、播放列表报真实时长。"""
import asyncio
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from peach.mp4index import (
composition_offsets_present, keyframe_seconds, movie_seconds, segment_plan,
)
from peach.hardware_transcode import HardwareEncoder
from peach.segments import HlsSegmentService, StreamSessionRegistry, build_hls_playlist
from support.mp4 import minimal_mp4
class Mp4IndexTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
def tearDown(self):
self.tmp.cleanup()
def write(self, payload: bytes, name="clip.mp4") -> Path:
path = self.root / name
path.write_bytes(payload)
return path
def test_keyframe_times_come_from_the_sync_sample_table(self):
# 1000 刻度、每样本 40 刻度（25fps），每 25 个样本一个关键帧 = 每秒一个。
path = self.write(minimal_mp4(timescale=1000, sample_delta=40, samples=250, keyframe_every=25))
self.assertEqual(keyframe_seconds(path)[:5], [0.0, 1.0, 2.0, 3.0, 4.0])
def test_moov_at_the_end_is_still_found(self):
"""非 faststart 的文件把 moov 放在结尾；顺序读到它可能要拉几个 GB。"""
full = minimal_mp4(timescale=1000, sample_delta=40, samples=100, keyframe_every=25)
moov_start = full.index(b"moov") - 4
moov_size = int.from_bytes(full[moov_start:moov_start + 4], "big")
moov = full[moov_start:moov_start + moov_size]
rest = full[:moov_start] + full[moov_start + moov_size:]
self.assertEqual(keyframe_seconds(self.write(rest + moov))[:3], [0.0, 1.0, 2.0])
def test_unparsable_input_returns_none_so_callers_fall_back_to_range(self):
self.assertIsNone(keyframe_seconds(self.write(b"not an mp4 at all")))
self.assertIsNone(keyframe_seconds(self.root / "missing.mp4"))
def test_composition_offsets_are_read_from_the_video_track(self):
"""有 B 帧却没有 ctts 的片源要改走转码，所以这张表在不在必须答得准。"""
plain = self.write(minimal_mp4(
timescale=1000, sample_delta=40, samples=100, keyframe_every=25))
reordered = self.write(minimal_mp4(
timescale=1000, sample_delta=40, samples=100, keyframe_every=25,
composition_offsets=True), name="reordered.mp4")
self.assertIs(composition_offsets_present(plain), False)
self.assertIs(composition_offsets_present(reordered), True)
def test_unreadable_moov_answers_none_instead_of_guessing(self):
"""读不出 moov 就不知道有没有这张表；猜「没有」会把好片源推去转码。"""
self.assertIsNone(composition_offsets_present(self.write(b"not an mp4 at all")))
self.assertIsNone(composition_offsets_present(self.root / "missing.mp4"))
def test_movie_duration_is_read_from_a_truncated_file_head(self):
"""远端视频只取文件头：`moov` 被截在一半也读得出 `mvhd`，放在结尾的读不出。"""
full = minimal_mp4(timescale=1000, sample_delta=40, samples=250, keyframe_every=25)
head = full[:full.index(b"mvhd") + 120]
self.assertEqual(movie_seconds(head), 10.0)
moov_start = full.index(b"moov") - 4
moov_size = int.from_bytes(full[moov_start:moov_start + 4], "big")
tail_moov = full[:moov_start] + full[moov_start + moov_size:] + full[moov_start:][:moov_size]
self.assertIsNone(movie_seconds(tail_moov[:len(head)]))
self.assertIsNone(movie_seconds(b"<html>not a video</html>"))
def test_plan_snaps_to_keyframes_and_covers_the_whole_duration(self):
keyframes = [round(value * 8.33, 2) for value in range(12)]
plan = segment_plan(keyframes, 100.0, 6)
self.assertTrue(all(start in keyframes for start, _ in plan),
"每段起点都必须落在真实关键帧上，否则 -c copy 切不准")
self.assertTrue(all(length >= 6 for _, length in plan[:-1]))
self.assertAlmostEqual(plan[0][0], 0.0)
self.assertAlmostEqual(plan[-1][0] + plan[-1][1], 100.0, places=6)
def test_plan_starts_at_zero_even_when_the_first_keyframe_is_late(self):
plan = segment_plan([3.0, 9.0, 15.0], 20.0, 6)
self.assertAlmostEqual(plan[0][0], 0.0)
self.assertAlmostEqual(plan[-1][0] + plan[-1][1], 20.0, places=6)
def test_playlist_reports_real_segment_durations(self):
"""按 6 秒等分声明会和关键帧实际切点对不上，播放器进度条就错位。"""
plan = [(0.0, 8.33), (8.33, 8.33), (16.66, 3.34)]
playlist = build_hls_playlist(plan, lambda index: f"/s/{index}.ts")
self.assertIn("#EXTINF:8.330,", playlist)
self.assertIn("#EXTINF:3.340,", playlist)
self.assertIn("#EXT-X-TARGETDURATION:8", playlist)
self.assertNotIn("#EXTINF:6.000,", playlist)
self.assertEqual(playlist.count("/s/"), 3)
self.assertIn("#EXT-X-ENDLIST", playlist)
if __name__ == "__main__":
unittest.main()
class SegmentCommandTests(unittest.TestCase):
"""片段的时间窗必须写成绝对终点，否则只有开头两段能产出。
真实故障：`-copyts` 保留原始时间轴后，`-t` 被 FFmpeg 当成绝对结束时刻而非片段时长。
每段的 `-t` 都约等于一个片段长，于是起点超过它的片段一律「已经过期」，FFmpeg 以
退出码 0、空 stderr 写出 0 字节，服务端只报一句没有内容的 ffmpeg failed。
实测 asset 6562：片段 0 与 1 返回 200，片段 2 和 300 全是 503，播到约 20 秒就断，
也拖不动。缓存目录里三个资产都只留下 0.ts 和 1.ts，没有更靠后的片段。
"""
def _command_for(self, index: int, transcode: bool = False) -> list[str]:
captured: list[list[str]] = []
class _Process:
returncode = 0
async def communicate(self):
return b"", b""
async def fake_exec(*command, **_kwargs):
captured.append(list(command))
return _Process()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "clip.mp4"
source.write_bytes(bytes(1024))
service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: type("C", (), {"path": "ffmpeg"})),
work_root=root / "cache",
)
# 计划固定为等长片段，免得测试依赖真实关键帧表。
service._plans[service.fingerprint(source)] = [
(round(n * 9.993, 3), 9.993) for n in range(400)
]
async def drive():
with mock.patch("asyncio.create_subprocess_exec", fake_exec):
# 产物由假进程负责写，否则会走到「0 字节」的失败分支。
original = service.cached_path
def cached_path(*args, **kwargs):
target = original(*args, **kwargs)
target.parent.mkdir(parents=True, exist_ok=True)
return target
with mock.patch.object(service, "cached_path", cached_path):
try:
await service.generate(
source, round(index * 9.993, 3), 9.993,
asset_id=6562, index=index, session="s",
registry=StreamSessionRegistry(),
transcode=transcode,
)
except Exception:
pass
asyncio.run(drive())
self.assertTrue(captured, "没有真的调用 FFmpeg")
return captured[0]
def test_conversion_slice_has_bounded_duration_and_absolute_output_timeline(self):
command = self._command_for(100, transcode=True)
self.assertEqual(command[command.index('-c:v')+1], 'libx264')
self.assertEqual(command[command.index('-t')+1], '9.993')
self.assertEqual(command[command.index('-output_ts_offset')+1], '999.300')
self.assertNotIn('-copyts', command)
def test_silent_ffmpeg_failure_still_says_what_was_observed(self):
"""FFmpeg 可以退出码 0、stderr 全空却写出 0 字节，只报 ffmpeg failed 等于没报。"""
from peach.segments import SegmentUnavailable
class _Process:
returncode = 0
async def communicate(self):
return b"", b""
async def fake_exec(*_command, **_kwargs):
return _Process()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "clip.mp4"
source.write_bytes(bytes(1024))
service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: type("C", (), {"path": "ffmpeg"})),
work_root=root / "cache",
)
async def drive():
with mock.patch("asyncio.create_subprocess_exec", fake_exec):
await service.generate(
source, 2997.995, 9.993, asset_id=6562, index=300,
session="s", registry=StreamSessionRegistry(),
)
with self.assertRaises(SegmentUnavailable) as caught:
asyncio.run(drive())
message = str(caught.exception)
self.assertIn("returncode=0", message)
self.assertIn("2997.995", message)
self.assertIn("3007.988", message)
def test_every_segment_asks_for_an_absolute_end_not_a_length(self):
for index in (0, 1, 2, 300):
command = self._command_for(index)
self.assertIn("-copyts", command)
self.assertNotIn("-t", command, f"片段 {index} 仍在用相对时长")
start = float(command[command.index("-ss") + 1])
end = float(command[command.index("-to") + 1])
self.assertAlmostEqual(end - start, 9.993, places=2)
# 决定性判据：终点随片段前移，而不是所有片段都停在第一段的末尾。
self.assertGreater(end, start)
self.assertAlmostEqual(end, round(index * 9.993, 3) + 9.993, places=2)
def test_concurrent_requests_generate_one_copy_of_the_same_segment(self):
"""播放器预取与重连会同时要同一段：同目标只许一个 FFmpeg，其余等它写完直接用缓存。"""
calls = 0
release = None
class _Process:
returncode = 0
async def communicate(self):
await release.wait()
return b"", b""
async def fake_exec(*command, **_kwargs):
nonlocal calls
calls += 1
Path(command[-1]).write_bytes(b"segment")
return _Process()
async def drive():
nonlocal release
release = asyncio.Event()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "clip.mp4"
source.write_bytes(bytes(1024))
service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: type("C", (), {"path": "ffmpeg"})),
work_root=root / "cache",
)
registry = StreamSessionRegistry()
with mock.patch("asyncio.create_subprocess_exec", fake_exec):
tasks = [asyncio.create_task(service.generate(
source, 0, 6, asset_id=6562, index=0, session=f"s{n}", registry=registry,
)) for n in range(3)]
await asyncio.sleep(0)
release.set()
results = await asyncio.gather(*tasks)
self.assertEqual(len({str(item) for item in results}), 1)
self.assertEqual(results[0].read_bytes(), b"segment")
self.assertEqual(service._generation_locks, {}, "锁在没人用时要从字典里消失")
asyncio.run(drive())
self.assertEqual(calls, 1)
def test_source_stats_and_cache_eviction_stay_off_the_event_loop(self):
"""片源 stat 可能卡在网盘上，淘汰要把缓存目录列一遍：都不在事件循环上做，淘汰按新写入字节攒够才跑。"""
on_loop, evictions = [], []
def running() -> bool:
try:
asyncio.get_running_loop()
except RuntimeError:
return False
return True
class _Process:
returncode = 0
async def communicate(self):
return b"", b""
async def fake_exec(*command, **_kwargs):
Path(command[-1]).write_bytes(bytes(100))
return _Process()
async def drive():
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "clip.mp4"
source.write_bytes(b"x")
# 上限 4000 字节，攒够十六分之一（250）才淘汰：每片 100，第三片之后跑一次。
service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: type("C", (), {"path": "ffmpeg"})),
work_root=root / "cache", cache_bytes=4000,
)
fingerprint, evict = service.fingerprint, service._evict
service.fingerprint = lambda path: (on_loop.append(running()), fingerprint(path))[1]
service._evict = lambda: (evictions.append(running()), evict())[1]
registry = StreamSessionRegistry()
with mock.patch("asyncio.create_subprocess_exec", fake_exec):
for index in range(5):
await service.generate(source, index * 6, 6, asset_id=1, index=index,
session="s", registry=registry)
await service.generate(source, 0, 6, asset_id=1, index=0, session="s", registry=registry)
asyncio.run(drive())
self.assertTrue(on_loop)
self.assertNotIn(True, on_loop, "片源 stat 不能在事件循环上")
self.assertEqual(evictions, [False])
def test_hardware_encode_failure_falls_back_to_libx264_for_the_slice(self):
"""没有 NVENC 的机器分片不能缺席：CUDA 解码、软件解码两条 NVENC 路都失败后走 libx264。"""
captured: list[list[str]] = self._transcode_commands("h264")
self.assertEqual([c[c.index("-c:v") + 1] for c in captured], ["h264_nvenc", "h264_nvenc", "libx264"])
self.assertIn("-hwaccel", captured[0])
self.assertNotIn("-hwaccel", captured[1])
for command in captured:
self.assertEqual(command[command.index("-output_ts_offset") + 1], "12.000")
self.assertEqual(command[command.index("-t") + 1], "6.000")
self.assertEqual(command[command.index("-f") + 1], "mpegts")
def test_a_codec_nvdec_garbles_is_decoded_in_software(self):
"""VC-1 交给 NVDEC 会解出绿帧，而 FFmpeg 退出码 0、stderr 全空，事后查不出来。
实测 `MIAD573_02.wmv`（vc1、1080p）：`-hwaccel cuda` 出来的首个分片开头 21 帧
在黑场与纯绿之间交替，同一段软件解码后色度全程正常。名单之外一律软件解码，
编码探测不出来的也是——省下的 CPU 抵不上一段没人察觉的坏画面。
"""
for codec in ("vc1", "wmv3", ""):
with self.subTest(codec=codec):
captured = self._transcode_commands(codec)
self.assertEqual([c[c.index("-c:v") + 1] for c in captured], ["h264_nvenc", "libx264"])
self.assertNotIn("-hwaccel", captured[0])
def _transcode_commands(self, codec: str) -> list[list[str]]:
"""NVENC 全程失败时，这个编码的片源实际跑过的命令。"""
captured: list[list[str]] = []
class _Process:
def __init__(self, returncode):
self.returncode = returncode
async def communicate(self):
return b"", (b"nvenc unavailable" if self.returncode else b"")
async def fake_exec(*command, **_kwargs):
captured.append(list(command))
if "h264_nvenc" in command:
return _Process(1)
Path(command[-1]).write_bytes(b"segment")
return _Process(0)
async def drive():
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "clip.mp4"
source.write_bytes(bytes(1024))
service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: type("C", (), {"path": "ffmpeg"})),
work_root=root / "cache", prefer_hardware=True,
)
with mock.patch("asyncio.create_subprocess_exec", fake_exec), \
mock.patch("peach.hardware_transcode.HardwareTranscoder._candidates",
return_value=(HardwareEncoder("nvenc"),)), \
mock.patch("peach.hardware_transcode.HardwareTranscoder._probe", return_value=True), \
mock.patch("peach.segments._probe_video_codec", return_value=codec):
target = await service.generate(
source, 12.0, 6.0, asset_id=6562, index=2, session="s",
registry=StreamSessionRegistry(), transcode=True,
)
self.assertEqual(target.read_bytes(), b"segment")
asyncio.run(drive())
return captured
class PlanCacheTests(unittest.TestCase):
"""计划缓存一满就 clear() 等于没有缓存；改成 LRU 后热条目要能活下来。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
self.service = HlsSegmentService(
resolver=mock.Mock(ffmpeg=lambda: None), work_root=self.root / "cache")
def tearDown(self):
self.tmp.cleanup()
def _source(self, number: int) -> Path:
path = self.root / f"clip{number}.mp4"
path.write_bytes(b"x")
return path
def _plan(self, number: int) -> None:
with mock.patch("peach.segments.keyframe_seconds", return_value=[0.0]):
self.service.plan(self._source(number), 60.0)
def _held_sources(self) -> set[str]:
return {key[0] for key in self.service._plans}
def test_cache_keeps_at_most_the_limit_and_evicts_the_oldest_access(self):
from peach.segments import PLAN_CACHE_LIMIT
for number in range(PLAN_CACHE_LIMIT):
self._plan(number)
self.assertEqual(len(self.service._plans), PLAN_CACHE_LIMIT)
# 命中一次让 0 号变成最近访问，再插入新条目：被逐出的必须是 1 号。
self._plan(0)
self._plan(PLAN_CACHE_LIMIT)
self.assertEqual(len(self.service._plans), PLAN_CACHE_LIMIT)
self.assertIn(str(self._source(0)), self._held_sources())
self.assertNotIn(str(self._source(1)), self._held_sources())