diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_stream_hevc_transcode.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_stream_hevc_transcode.py | 155 |
1 files changed, 0 insertions, 155 deletions
diff --git a/packages/meshbay-node/tests/test_stream_hevc_transcode.py b/packages/meshbay-node/tests/test_stream_hevc_transcode.py deleted file mode 100644 index 7d831ee..0000000 --- a/packages/meshbay-node/tests/test_stream_hevc_transcode.py +++ /dev/null @@ -1,155 +0,0 @@ -""" -HEVC video is transcoded to H264 for streaming, never copied — unlike the -codecs media_probe.py's BROWSER_INCOMPATIBLE_VIDEO_CODECS excludes. - -Found live: a real HEVC/EAC3 WEB-DL streamed fine over MNP (ffprobe/VLC play -it) but the browser reported "Codec not supported for streaming: -hev1.1.6.L93.B0,mp4a.40.2" from MediaSource.isTypeSupported — Chrome has no -HEVC decoder on most non-Apple platforms. "-c:v copy" on an incompatible -codec is not a mux failure the way EAC3 audio is (test_stream_audio_ -transcode.py); ffmpeg happily remuxes it, and the browser is the one that -then refuses it, silently, at playback rather than at stream_init. - -These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi -test sources, ~1s), the same style as test_stream_audio_transcode.py. -""" - -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, _probe_video - -from conftest import needs_subprocess, one_root - -_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") -_HAVE_HEVC_ENCODER = _HAVE_FFMPEG and b"libx265" in subprocess.run( - ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True).stdout -pytestmark = [ - pytest.mark.asyncio, - pytest.mark.skipif(not _HAVE_HEVC_ENCODER, reason="ffmpeg/libx265 not installed"), - needs_subprocess, -] - - -def _make_hevc_clip(path: Path) -> None: - """~1s of HEVC video + AAC audio — a minimal stand-in for a real HEVC WEB-DL.""" - 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", - "-c:v", "libx265", "-preset", "ultrafast", "-c:a", "aac", - str(path)], - check=True, capture_output=True, - ) - - -def _session(video_path: Path, gek: bytes, *, transcode_incompatible_video: bool = True): - 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, - "transcode_incompatible_video": transcode_incompatible_video, - } - 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 - - -def _output_video_codec(path: Path) -> str: - probe = subprocess.run( - ["ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream=codec_name", "-of", "csv=p=0", str(path)], - check=True, capture_output=True, text=True) - return probe.stdout.strip() - - -async def test_hevc_video_is_transcoded_to_h264_by_default(tmp_path): - clip = tmp_path / "clip.mkv" - _make_hevc_clip(clip) - gek = generate_gek() - session, file_id = _session(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 init["codec"].startswith("avc1."), ( - "the reported codec must be the transcoded H264 string, never the " - f"source's hev1 string a browser cannot decode: {init['codec']}") - - remuxed = _reassemble(session.sent, gek, file_id) - out_path = tmp_path / "out.mp4" - out_path.write_bytes(remuxed) - assert _output_video_codec(out_path) == "h264", \ - "the bytes on the wire must actually be H264, not just the reported label" - - -async def test_hevc_transcode_can_be_disabled_by_the_operator(tmp_path): - clip = tmp_path / "clip.mkv" - _make_hevc_clip(clip) - gek = generate_gek() - session, file_id = _session(clip, gek, transcode_incompatible_video=False) - - 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"] - init = next(m for m in session.sent if m.get("type") == "stream_init") - assert init["codec"].startswith("hev1."), ( - "with the fallback disabled, the source is copied as-is and the " - f"original HEVC codec string must be reported unchanged: {init['codec']}") - - remuxed = _reassemble(session.sent, gek, file_id) - out_path = tmp_path / "out.mp4" - out_path.write_bytes(remuxed) - assert _output_video_codec(out_path) == "hevc", \ - "with the fallback disabled, the wire bytes must still be copied HEVC" - - -async def test_probe_video_reports_raw_codec_name_for_hevc(tmp_path): - clip = tmp_path / "clip.mkv" - _make_hevc_clip(clip) - - codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) - - assert raw_codec == "hevc" - assert codec is not None and codec.startswith("hev1.") |