add more tests

tests/test_filters.py

@@ -0,0 +1,124 @@
+"""
+test classes and functions in filters/default.py
+"""
+
+from types import SimpleNamespace
+
+from ..ffplayout.filters.default import (add_audio, add_loudnorm,
+                                         deinterlace_filter, extend_audio,
+                                         extend_video, fade_filter, fps_filter,
+                                         overlay_filter, pad_filter,
+                                         realtime_filter, scale_filter,
+                                         split_filter, text_filter)
+from ..ffplayout.utils import lower_third, pre, sync_op
+
+
+def test_text_filter():
+    lower_third.add_text = True
+    lower_third.over_pre = True
+    lower_third.address = '127.0.0.1:5555'
+    lower_third.fontfile = ''
+
+    assert text_filter() == [
+        "null,zmq=b=tcp\\\\://'127.0.0.1\\:5555',drawtext=text=''"]
+
+
+def test_deinterlace_filter():
+    probe = SimpleNamespace(video=[{'field_order': 'tff'}])
+
+    assert deinterlace_filter(probe) == ['yadif=0:-1:0']
+
+
+def test_pad_filter():
+    probe = SimpleNamespace(video=[{'aspect': 1.333}])
+
+    assert pad_filter(probe) == ['pad=ih*1024/576/sar:ih:(ow-iw)/2:(oh-ih)/2']
+
+
+def test_fps_filter():
+    probe = SimpleNamespace(video=[{'fps': 29.97}])
+
+    assert fps_filter(probe) == ['fps=25']
+
+
+def test_scale_filter():
+    probe = SimpleNamespace(video=[{'width': 1440, 'height': 1080,
+                                    'aspect': 1.333}])
+
+    assert scale_filter(probe) == ['scale=1024:576', 'setdar=dar=1.778']
+
+
+def test_fade_filter():
+    assert fade_filter(300, 5, 300) == ['fade=in:st=0:d=0.5']
+    assert fade_filter(300, 5, 300, 'a') == ['afade=in:st=0:d=0.5']
+    assert fade_filter(300, 0, 200) == ['fade=out:st=199.0:d=1.0']
+    assert fade_filter(300, 0, 200, 'a') == ['afade=out:st=199.0:d=1.0']
+
+
+def test_overlay_filter():
+    assert overlay_filter(300, True, False, False) == '[v]null'
+    assert overlay_filter(300, False, True, False) == (
+        'movie=docs/logo.png,loop=loop=-1:size=1:start=0,format=rgba,'
+        'colorchannelmixer=aa=0.7,fade=in:st=0:d=1.0:alpha=1[l];'
+        '[v][l]overlay=W-w-12:12:shortest=1')
+    assert overlay_filter(300, False, False, True) == (
+        'movie=docs/logo.png,loop=loop=-1:size=1:start=0,format=rgba,'
+        'colorchannelmixer=aa=0.7,fade=out:st=299:d=1.0:alpha=1[l];'
+        '[v][l]overlay=W-w-12:12:shortest=1')
+    assert overlay_filter(300, False, False, False) == (
+        'movie=docs/logo.png,loop=loop=-1:size=1:start=0,format=rgba,'
+        'colorchannelmixer=aa=0.7[l];[v][l]overlay=W-w-12:12:shortest=1')
+
+
+def test_add_audio():
+    probe = SimpleNamespace(audio=False, src='/path/file.mp4')
+
+    assert add_audio(probe, 300) == [
+        ('aevalsrc=0:channel_layout=stereo:duration=300:sample_rate=48000')]
+
+
+def test_add_loudnorm():
+    pre.add_loudnorm = True
+    pre.loud_i = -18
+    pre.loud_tp = -1.5
+    pre.loud_lra = 11
+    probe = SimpleNamespace(audio=True)
+
+    assert add_loudnorm(probe) == ['loudnorm=I=-18:TP=-1.5:LRA=11']
+
+
+def test_extend_audio():
+    probe = SimpleNamespace(audio=[{'duration': 299}])
+
+    assert extend_audio(probe, 300, 0) == ['apad=whole_dur=300']
+    assert extend_audio(probe, 300, 10) == ['apad=whole_dur=290']
+
+
+def test_extend_video():
+    probe = SimpleNamespace(video=[{'duration': 299}])
+
+    assert extend_video(probe, 300, 0) == [
+        'tpad=stop_mode=add:stop_duration=1.0']
+
+    assert extend_video(probe, 300, 10) == [
+        'tpad=stop_mode=add:stop_duration=1.0']
+
+
+def test_realtime_filter():
+    sync_op.realtime = True
+
+    assert realtime_filter(300) == ',realtime=speed=1'
+    assert realtime_filter(300, 'a') == ',arealtime=speed=1'
+
+    sync_op.time_delta = -1.0
+
+    assert realtime_filter(300) == ',realtime=speed=1.0033444816053512'
+
+
+def test_split_filter():
+    pre.output_count = 1
+    assert split_filter('v') == '[vout1]'
+
+    pre.output_count = 3
+    assert split_filter('v') == ',split=3[vout1][vout2][vout3]'
+    assert split_filter('a') == ',asplit=3[aout1][aout2][aout3]'

tests/test_playlist.py

@@ -0,0 +1,88 @@
+"""
+test classes and functions in playlist.py
+"""
+
+import datetime
+from zoneinfo import ZoneInfo
+
+import time_machine
+
+from ..ffplayout.playlist import (handle_list_end, handle_list_init,
+                                  timed_source)
+from ..ffplayout.utils import playlist, storage
+
+# set time zone
+_TZ = ZoneInfo("Europe/Berlin")
+
+
+@time_machine.travel(datetime.datetime(*[2021, 5, 31, 6, 0, 20], tzinfo=_TZ))
+def test_handle_list_init():
+    playlist.start = 6 * 60 * 60
+    storage.filler = ''
+    node = {'source': '/store/file.mp4', 'begin': 21620,
+            'in': 0, 'out': 300, 'duration': 300, 'seek': 20}
+
+    color_src = ('color=c=#121212:s=1024x576:d=280.0:r=25,'
+                 'format=pix_fmts=yuv420p')
+
+    list_init = handle_list_init(node)
+    list_init.pop('probe')
+    check_result = {
+        'source': color_src,
+        'begin': 21620, 'in': 0, 'out': 300,
+        'duration': 300, 'seek': 20.0,
+        'src_cmd': [
+            '-f', 'lavfi', '-i', color_src,
+            '-f', 'lavfi', '-i', 'anoisesrc=d=280.0:c=pink:r=48000:a=0.05']}
+
+    assert list_init == check_result
+
+
+@time_machine.travel(datetime.datetime(*[2021, 5, 31, 5, 59, 30], tzinfo=_TZ))
+def test_handle_list_end():
+    playlist.start = 6 * 60 * 60
+    storage.filler = ''
+
+    node = {'source': '/store/file.mp4', 'begin': 24 * 3600 - 30,
+            'in': 0, 'out': 300, 'duration': 300, 'seek': 0}
+
+    color_src = ('color=c=#121212:s=1024x576:d=30:r=25,'
+                 'format=pix_fmts=yuv420p')
+
+    check_result = {
+        'source': color_src,
+        'begin': 24 * 3600 - 30, 'in': 0, 'out': 30,
+        'duration': 300, 'seek': 0,
+        'src_cmd': [
+            '-f', 'lavfi', '-i', color_src,
+            '-f', 'lavfi', '-i', 'anoisesrc=d=30:c=pink:r=48000:a=0.05']}
+
+    list_end = handle_list_end(30, node)
+    list_end.pop('probe')
+
+    assert list_end == check_result
+
+
+@time_machine.travel(datetime.datetime(*[2021, 5, 31, 5, 50, 00], tzinfo=_TZ))
+def test_timed_source():
+    playlist.start = 6 * 60 * 60
+    storage.filler = ''
+
+    node = {'source': '/store/file.mp4', 'begin': 24 * 3600 + 21600 - 600,
+            'in': 0, 'out': 300, 'duration': 300, 'seek': 0}
+
+    color_src = ('color=c=#121212:s=1024x576:d=300:r=25,'
+                 'format=pix_fmts=yuv420p')
+
+    check_result = {
+        'source': color_src,
+        'begin': 24 * 3600 + 21600 - 600, 'in': 0, 'out': 300,
+        'duration': 300, 'seek': 0,
+        'src_cmd': [
+            '-f', 'lavfi', '-i', color_src,
+            '-f', 'lavfi', '-i', 'anoisesrc=d=300:c=pink:r=48000:a=0.05']}
+
+    src = timed_source(node, False)
+    src.pop('probe')
+
+    assert src == check_result