63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from restreamer import ffmpeg_runner
|
|
from restreamer.constants import FFMPEG_TERMINATE_TIMEOUT_SECONDS
|
|
from restreamer.ffmpeg_runner import FfmpegStreamProcess, build_ffmpeg_args
|
|
from restreamer.resolver import ResolvedStream
|
|
|
|
|
|
def test_build_ffmpeg_args_single_input_without_headers() -> None:
|
|
stream = ResolvedStream(video_url="https://example.com/stream.m3u8")
|
|
|
|
args = build_ffmpeg_args(stream)
|
|
|
|
assert args[0] == "ffmpeg"
|
|
assert "-headers" not in args
|
|
assert args.count("-i") == 1
|
|
assert stream.video_url in args
|
|
assert args[-1] == "pipe:1"
|
|
|
|
|
|
def test_build_ffmpeg_args_includes_headers_when_present() -> None:
|
|
stream = ResolvedStream(
|
|
video_url="https://example.com/stream.m3u8",
|
|
http_headers={"User-Agent": "test-agent", "Cookie": "a=b"},
|
|
)
|
|
|
|
args = build_ffmpeg_args(stream)
|
|
|
|
assert "-headers" in args
|
|
header_value = args[args.index("-headers") + 1]
|
|
assert "User-Agent: test-agent\r\n" in header_value
|
|
assert "Cookie: a=b\r\n" in header_value
|
|
|
|
|
|
def test_build_ffmpeg_args_maps_separate_video_and_audio_inputs() -> None:
|
|
stream = ResolvedStream(
|
|
video_url="https://example.com/video.m3u8",
|
|
audio_url="https://example.com/audio.m3u8",
|
|
)
|
|
|
|
args = build_ffmpeg_args(stream)
|
|
|
|
assert args.count("-i") == 2
|
|
assert stream.video_url in args
|
|
assert stream.audio_url in args
|
|
assert "-map" in args
|
|
map_values = [args[i + 1] for i, a in enumerate(args) if a == "-map"]
|
|
assert map_values == ["0:v:0", "1:a:0"]
|
|
|
|
|
|
async def test_stop_returns_when_consumer_left_stdout_unread(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(ffmpeg_runner, "build_ffmpeg_args", lambda stream: ["yes"])
|
|
process = FfmpegStreamProcess(ResolvedStream(video_url="https://example.com/stream.m3u8"), "test")
|
|
await process.start()
|
|
async for _chunk in process.iter_chunks():
|
|
break
|
|
|
|
await asyncio.wait_for(process.stop(), timeout=FFMPEG_TERMINATE_TIMEOUT_SECONDS / 2)
|
|
|
|
assert await process.wait() != 0
|