119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
import pytest
|
|
|
|
from recorder import RecordingManager, validate_ffmpeg_args, validate_stream_url, PROTOCOL_WHITELIST
|
|
from scheduler import compute_duration, validate_schedule
|
|
|
|
|
|
@pytest.mark.parametrize("args", [
|
|
"",
|
|
"-c:a libmp3lame -b:a 128k",
|
|
"-metadata title=Foo/Bar",
|
|
"-metadata 'title=Mit Leerzeichen'",
|
|
"-vn -ar 44100 -ac 2",
|
|
"-af volume=2",
|
|
])
|
|
def test_ffmpeg_args_allowed(args):
|
|
validate_ffmpeg_args(args)
|
|
|
|
|
|
@pytest.mark.parametrize("args", [
|
|
"/app/app.py",
|
|
"-i x",
|
|
"-af amovie=/etc/passwd",
|
|
"-filter:a AMOVIE=x",
|
|
"-af sendcmd=f=x",
|
|
"-c:a",
|
|
"-c:a -b:a",
|
|
"-b:a 128k out.mp3",
|
|
"-y",
|
|
"-f segment -segment_list /tmp/x",
|
|
"-metadata ''",
|
|
"'unclosed",
|
|
"-vn " * 200,
|
|
])
|
|
def test_ffmpeg_args_rejected(args):
|
|
with pytest.raises(ValueError):
|
|
validate_ffmpeg_args(args)
|
|
|
|
|
|
def test_ffmpeg_args_error_names_token():
|
|
with pytest.raises(ValueError, match="/app/app.py"):
|
|
validate_ffmpeg_args("-vn /app/app.py")
|
|
|
|
|
|
@pytest.mark.parametrize("url", [
|
|
"https://example.com/stream.mp3",
|
|
"http://radio.example:8000/live",
|
|
"rtmp://live.example/app/key",
|
|
"rtsps://cam.example/stream",
|
|
])
|
|
def test_url_allowed(url):
|
|
assert validate_stream_url(url) == url
|
|
|
|
|
|
@pytest.mark.parametrize("url", [
|
|
"file:///etc/passwd",
|
|
"concat:/etc/passwd|/etc/shadow",
|
|
"subfile:,,start,0,end,0,,:/etc/passwd",
|
|
"https://",
|
|
"http://exa mple.com/",
|
|
"https://example.com/\nfoo",
|
|
"ftp://example.com/x",
|
|
"",
|
|
"https://example.com/" + "a" * 1000,
|
|
])
|
|
def test_url_rejected(url):
|
|
with pytest.raises(ValueError):
|
|
validate_stream_url(url)
|
|
|
|
|
|
def test_ffmpeg_cmd_has_protocol_whitelist_before_input():
|
|
mgr = RecordingManager(config=None)
|
|
cmd = mgr._build_ffmpeg_cmd("https://example.com/a.m3u8", "hls", "/x/out.mp4",
|
|
"-metadata title=X", False, None)
|
|
assert cmd.index("-protocol_whitelist") < cmd.index("-i")
|
|
assert cmd[cmd.index("-protocol_whitelist") + 1] == PROTOCOL_WHITELIST
|
|
assert "file" not in PROTOCOL_WHITELIST.split(",")
|
|
assert cmd[-1] == "/x/out.mp4"
|
|
assert cmd[-3:-1] == ["-metadata", "title=X"]
|
|
|
|
|
|
def test_ffmpeg_cmd_revalidates():
|
|
mgr = RecordingManager(config=None)
|
|
with pytest.raises(ValueError):
|
|
mgr._build_ffmpeg_cmd("https://example.com/a", "http", "/x/out.mp4", "/app/app.py", False, None)
|
|
with pytest.raises(ValueError):
|
|
mgr._build_ffmpeg_cmd("file:///etc/passwd", "http", "/x/out.mp4", "", False, None)
|
|
|
|
|
|
@pytest.mark.parametrize("job,msg", [
|
|
({"schedule_start": "25:00"}, "HH:MM"),
|
|
({"schedule_start": "10:00", "schedule_stop": "9:5"}, "HH:MM"),
|
|
({"schedule_date": "2026-02-30"}, "JJJJ"),
|
|
({"schedule_days": "mo-xx"}, "Wochentage"),
|
|
({"schedule_enabled": True}, "Startzeit"),
|
|
({"schedule_once": True, "schedule_start": "10:00"}, "Datum"),
|
|
({"schedule_date": "2026-05-02", "schedule_end_date": "2026-05-01"}, "Enddatum"),
|
|
])
|
|
def test_schedule_invalid(job, msg):
|
|
with pytest.raises(ValueError, match=msg):
|
|
validate_schedule(job)
|
|
|
|
|
|
def test_schedule_valid():
|
|
validate_schedule({"schedule_enabled": True, "schedule_start": "23:30", "schedule_stop": "01:00",
|
|
"schedule_days": "Mo-Fr,SA"})
|
|
validate_schedule({"schedule_once": True, "schedule_date": "2026-05-01",
|
|
"schedule_end_date": "2026-05-03", "schedule_start": "10:00"})
|
|
|
|
|
|
def test_compute_duration():
|
|
assert compute_duration({"max_duration": 120}, apply_buffers=True) == 120
|
|
assert compute_duration({"schedule_start": "10:00", "schedule_stop": "11:00"}, False) == 3600
|
|
assert compute_duration({"schedule_start": "23:00", "schedule_stop": "01:00"}, False) == 7200
|
|
assert compute_duration({"schedule_start": "10:00", "schedule_stop": "11:00",
|
|
"pre_buffer": 5, "post_buffer": 5}, True) == 4200
|
|
assert compute_duration({"schedule_start": "10:00", "schedule_stop": ""}, True) is None
|
|
with pytest.raises(ValueError):
|
|
compute_duration({"schedule_start": "10:00", "schedule_stop": "xx"}, False)
|