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
|
"""
Audio in a streamed video is always transcoded to AAC, never copied.
Found live against a real 5.1 E-AC-3 WEB-DL (season 2 of a show whose season 1
was AAC and played fine): "-c copy" on an E-AC-3 track makes ffmpeg itself
refuse to write the fragmented MP4 header at all — "Cannot write moov atom
before EAC3 packets parsed" — and even for the codecs that don't make ffmpeg
outright refuse (plain AC-3, DTS, ...), no mainstream browser's MSE decodes
them, so the viewer would see a stream that starts and immediately ends with
no picture and no error. Video is untouched here — it is always copied,
still H264 in every fixture below, since re-encoding it is the expensive
thing this pipeline exists to avoid.
These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi
test sources, ~1s) rather than asserting against the source text — a source
match cannot tell a working remux from one ffmpeg silently refuses to write.
"""
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.media_probe import probe_video as _probe_video
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.asyncio,
pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed"),
needs_subprocess,
]
def _make_clip(path: Path, *, acodec: str, channels: int = 2) -> None:
"""~1s of H264 video + `acodec` audio, muxed into an .mkv — a minimal
stand-in for a real WEB-DL with that audio codec."""
subprocess.run(
["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
"-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1",
"-f", "lavfi", "-i", "sine=frequency=440:duration=1:sample_rate=48000",
"-ac", str(channels),
"-c:v", "libx264", "-preset", "ultrafast", "-c:a", acodec,
str(path)],
check=True, capture_output=True,
)
def _session(tmp_path: Path, video_path: Path, gek: bytes) -> WebRTCPeerSession:
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)
segments = sorted(
(m for m in sent if m.get("type") == "stream_data"),
key=lambda m: m["segment_index"])
out = b""
for m in segments:
key = chunk_key_aes(gek, file_hash, m["segment_index"])
out += decrypt_chunk_aes(key, m["nonce"], m["ct"])
return out
async def test_eac3_audio_is_transcoded_not_copied(tmp_path):
"""The exact reproduction: ffmpeg refuses "-c copy" on E-AC-3 outright."""
clip = tmp_path / "clip.mkv"
_make_clip(clip, acodec="eac3", channels=6)
gek = generate_gek()
session, file_id = _session(tmp_path, clip, gek)
await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0})
errors = [m for m in session.sent if m.get("type") == "error"]
assert not errors, f"streaming must not fail: {errors}"
init = next(m for m in session.sent if m.get("type") == "stream_init")
assert "mp4a.40.2" in init["codec"], \
"the reported codec must be AAC, never the source's eac3"
data_segments = [m for m in session.sent if m.get("type") == "stream_data"]
assert data_segments, "no video data was ever sent"
assert any(m.get("type") == "stream_end" for m in session.sent)
remuxed = _reassemble(session.sent, gek, file_id)
out_path = tmp_path / "out.mp4"
out_path.write_bytes(remuxed)
probe = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "stream=codec_name,codec_type,channels",
"-of", "csv=p=0", str(out_path)],
check=True, capture_output=True, text=True)
rows = [line.split(",") for line in probe.stdout.strip().splitlines()]
codecs = {r[1]: r[0] for r in rows}
channels = {r[1]: r[2] for r in rows if len(r) > 2}
assert codecs.get("audio") == "aac", f"audio must be AAC on the wire: {codecs}"
assert codecs.get("video") == "h264", f"video must still be H264: {codecs}"
assert channels.get("audio") == "2", (
"audio must be downmixed to stereo: some browsers' MSE decoder "
f"silently rejects multichannel AAC once real fragments are "
f"appended, even though ffprobe/VLC accept it fine: {channels}")
async def test_already_aac_audio_still_streams(tmp_path):
"""Season-1-shaped file (already AAC): must keep working exactly as before."""
clip = tmp_path / "clip.mkv"
_make_clip(clip, acodec="aac", channels=2)
gek = generate_gek()
session, file_id = _session(tmp_path, clip, gek)
await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0})
assert not [m for m in session.sent if m.get("type") == "error"]
assert any(m.get("type") == "stream_data" for m in session.sent)
assert any(m.get("type") == "stream_end" for m in session.sent)
async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path):
clip = tmp_path / "clip.mkv"
_make_clip(clip, acodec="eac3", channels=6)
probe = await _probe_video(str(clip))
assert probe.has_audio is True
assert probe.duration > 0
assert probe.codec is not None
assert "eac3" not in probe.codec and "ec-3" not in probe.codec
assert "mp4a.40.2" in probe.codec
assert (probe.width, probe.height) == (320, 240)
assert probe.raw_codec_name == "h264"
async def test_probe_video_handles_no_audio_track(tmp_path):
clip = tmp_path / "silent.mkv"
subprocess.run(
["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
"-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1",
"-c:v", "libx264", "-preset", "ultrafast", "-an", str(clip)],
check=True, capture_output=True,
)
probe = await _probe_video(str(clip))
assert probe.has_audio is False
assert probe.audio_tracks == []
assert probe.codec is not None and "," not in probe.codec, \
"no audio track must not produce a dangling ',' or a fake audio codec"
assert (probe.width, probe.height) == (320, 240)
assert probe.raw_codec_name == "h264"
|