""" A seek on a copied stream reports where the picture actually begins. Copied video has to start on a keyframe, so `-ss t` with `-c:v copy` delivers the keyframe at or before `t` — up to a whole GOP earlier. The node reported `t` regardless, and the client sets `SourceBuffer.timestampOffset` from that number, so everything downstream believed the picture stood a few seconds further along than it did. That was a wrong label while only the scrubber read it. It became a wrong *answer* when subtitles arrived: their cues carry the source's own absolute timestamps, so the mismatch put every line on screen before it was spoken. **Where the seek lands is measured, not predicted, and that distinction is this module's subject.** The first attempt scanned ffprobe's key frames and took the last one at or before the request. It was wrong twice: Matroska's Cues index only some keyframes, so the seek backs off to an indexed one that can be much earlier, and the landing point moves with **which streams are mapped**, because the container is positioned where every mapped stream has data. On a real title, one seek answered 4909.863 s from the frame list and delivered 4907.236 s — 2.65 s of subtitles standing away from the voice, which is what a viewer reported after the first fix. **The assertion is therefore on the decoded picture, not on the number.** A test comparing `stream_init["start"]` against an expected keyframe would agree with the implementation by construction, both reading the same probe. The first frame delivered is decoded and matched against the source frame at the position the node claims, which is a statement about what was served rather than about what was computed. """ import shutil import subprocess from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.transport.webrtc.media_tools import _seek_lands_at from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root _HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") pytestmark = [ pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed"), needs_subprocess, ] # 25 fps with a keyframe every 250 frames — one every 10 s, which is what a # real WEB-DL looks like and what makes the gap large enough to see. _GOP_SECONDS = 10 _SEEK_TO = 15.0 # between the keyframes at 10 s and 20 s _EXPECTED_KEYFRAME = 10.0 def _make_h264_clip(path: Path) -> None: """30 s of H264 the streaming path will copy rather than re-encode. The picture has to differ from one second to the next, or a frame from the keyframe and a frame from the requested position would compare equal and the test would pass against the bug it exists for. """ subprocess.run( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=30", "-f", "lavfi", "-i", "sine=duration=30", "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-g", str(_GOP_SECONDS * 25), "-keyint_min", str(_GOP_SECONDS * 25), "-sc_threshold", "0", "-c:a", "aac", "-shortest", str(path)], check=True, capture_output=True) def _session(video_path: Path, gek: bytes): import blake3 file_bytes = video_path.read_bytes() file_id = blake3.blake3(file_bytes).hexdigest() sk_node = Ed25519PrivateKey.generate() index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek) from meshbay_common.protocol import IndexEntry index.add_entry(IndexEntry( id=file_id, name=video_path.name, path=video_path.parent.name, size=len(file_bytes), type="video", added_at=0)) session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = { "roots": one_root(video_path.parent), "index": index, "gek": gek, "sk_node": sk_node, "max_concurrent_streams": 4, } session._group_id = None session._user_id = "tester" session._stream_stopped = False session._stream_keepalives = 0 session.sent = [] session._send = session.sent.append session._audit = lambda *a, **k: None return session, file_id def _reassemble(sent: list[dict], gek: bytes, file_id: str) -> bytes: file_hash = bytes.fromhex(file_id) out = b"" for m in sorted((m for m in sent if m.get("type") == "stream_data"), key=lambda m: m["segment_index"]): key = chunk_key_aes(gek, file_hash, m["segment_index"]) out += decrypt_chunk_aes(key, m["nonce"], m["ct"]) return out def _frame_md5(path: Path, at: float | None = None) -> str: """One decoded frame as raw pixels, hashed. `at` None means the first.""" args = ["ffmpeg", "-hide_banner", "-loglevel", "error"] if at is not None: args += ["-ss", f"{at:.6f}"] args += ["-i", str(path), "-frames:v", "1", "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] proc = subprocess.run(args, check=True, capture_output=True) import hashlib return hashlib.md5(proc.stdout).hexdigest() _MAPS = ["-map", "0:v:0", "-map", "0:a:0"] @pytest.mark.asyncio async def test_the_probe_measures_where_the_seek_lands(tmp_path): """Measured with the mapping the stream will use, never predicted. A scan of ffprobe's key frames answers a different question: Matroska indexes only some keyframes, and the landing point also moves with which audio track is mapped, because the container is positioned where every mapped stream has data. On a real title those two answers were 4909.863 and 4907.236 — 2.65 s of subtitles standing away from the voice. """ clip = tmp_path / "clip.mp4" _make_h264_clip(clip) assert await _seek_lands_at(clip, _SEEK_TO, _MAPS) == pytest.approx( _EXPECTED_KEYFRAME, abs=0.05) # A position that *is* a keyframe answers itself, not the one before. assert await _seek_lands_at(clip, 20.0, _MAPS) == pytest.approx(20.0, abs=0.05) # A seek never lands after what was asked for. for t in (7.0, 15.0, 23.0): landed = await _seek_lands_at(clip, t, _MAPS) assert landed is not None and landed <= t + 0.05, (t, landed) @pytest.mark.asyncio async def test_a_seek_reports_where_the_picture_begins_not_where_it_was_asked(tmp_path): clip = tmp_path / "clip.mp4" _make_h264_clip(clip) gek = generate_gek() session, file_id = _session(clip, gek) await session._stream_video_inner( {"file_id": file_id, "start": _SEEK_TO, "credits": 0}) errors = [m for m in session.sent if m.get("type") == "error"] assert not errors, errors init = next(m for m in session.sent if m.get("type") == "stream_init") assert init["start"] == pytest.approx(_EXPECTED_KEYFRAME, abs=0.05), ( f"the node announced {init['start']}, but a copied stream cannot " f"begin anywhere but the keyframe at {_EXPECTED_KEYFRAME}") assert init["start"] < _SEEK_TO, ( "this fixture must have a keyframe gap to report, or the test cannot " "tell the fix from the fault") @pytest.mark.asyncio async def test_the_announced_position_is_the_picture_that_was_served(tmp_path): """Decoded and compared against the source — see this module's docstring.""" clip = tmp_path / "clip.mp4" _make_h264_clip(clip) gek = generate_gek() session, file_id = _session(clip, gek) await session._stream_video_inner( {"file_id": file_id, "start": _SEEK_TO, "credits": 0}) init = next(m for m in session.sent if m.get("type") == "stream_init") served = tmp_path / "served.mp4" served.write_bytes(_reassemble(session.sent, gek, file_id)) assert _frame_md5(served) == _frame_md5(clip, at=init["start"]), ( "the first frame delivered is not the source frame at the position " "the node announced, so `start` still does not name what was served") # And it is emphatically not the frame at the position asked for, or the # comparison above would hold for the old behaviour too. assert _frame_md5(served) != _frame_md5(clip, at=_SEEK_TO), ( "the fixture's picture does not change across the keyframe gap, so " "this comparison proves nothing")