1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
"""
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 a GOP before it was
spoken. Measured on a real H264 title, seeking to 600 s, 2650 s and 5000 s
landed on keyframes 0.82 s, 1.56 s and 4.64 s earlier.
**The assertion is on the decoded picture, not on the number.** A test that
only compared `stream_init["start"]` against an expected keyframe would agree
with the implementation by construction — both would be reading the same
ffprobe. 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_server import (
WebRTCPeerSession,
_keyframe_at_or_before,
)
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()
@pytest.mark.asyncio
async def test_the_lookup_finds_the_keyframe_the_seek_will_land_on(tmp_path):
clip = tmp_path / "clip.mp4"
_make_h264_clip(clip)
assert await _keyframe_at_or_before(clip, _SEEK_TO) == pytest.approx(
_EXPECTED_KEYFRAME, abs=0.05)
# A position that *is* a keyframe answers itself, not the one before.
assert await _keyframe_at_or_before(clip, 20.0) == pytest.approx(20.0, abs=0.05)
# Before the first one there is nothing earlier to find.
assert await _keyframe_at_or_before(clip, 0) == 0.0
@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")
|