|
1
|
|
|
import json |
|
2
|
|
|
import os |
|
3
|
|
|
import unittest |
|
4
|
|
|
|
|
5
|
|
|
import ffmpeg_streaming |
|
6
|
|
|
from ffmpeg_streaming import Formats |
|
7
|
|
|
from ffmpeg_streaming import FFProbe, Representation, Size, Bitrate |
|
8
|
|
|
from ffmpeg_streaming._media import HLS, DASH |
|
9
|
|
|
from ffmpeg_streaming.ffprobe import Streams |
|
10
|
|
|
|
|
11
|
|
|
|
|
12
|
|
|
class TestStreaming(unittest.TestCase): |
|
13
|
|
|
|
|
14
|
|
|
def setUp(self): |
|
15
|
|
|
self.src_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'files') |
|
16
|
|
|
self.src_video = os.path.join(self.src_dir, 'test.mp4') |
|
17
|
|
|
with open(os.path.join(self.src_dir, 'fixture_ffprobe')) as ffprobe: |
|
18
|
|
|
self.ffprobe = json.loads(ffprobe.read()) |
|
19
|
|
|
|
|
20
|
|
|
def test_ffprobe(self): |
|
21
|
|
|
ffprobe = FFProbe(self.src_video) |
|
22
|
|
|
|
|
23
|
|
|
all_media = ffprobe.all() |
|
24
|
|
|
|
|
25
|
|
|
streams = ffprobe.streams() |
|
26
|
|
|
self.assertIsInstance(streams, Streams) |
|
27
|
|
|
self.assertEqual(streams.all(), all_media['streams']) |
|
28
|
|
|
self.assertEqual(streams.first_stream(), all_media['streams'][0]) |
|
29
|
|
|
self.assertEqual(streams.video()['codec_type'], 'video') |
|
30
|
|
|
self.assertEqual(streams.audio()['codec_type'], 'audio') |
|
31
|
|
|
|
|
32
|
|
|
def test_hls(self): |
|
33
|
|
|
video = ffmpeg_streaming.input(self.src_video) |
|
34
|
|
|
hls_obj = video.hls(Formats.h264()) |
|
35
|
|
|
self.assertIsInstance(hls_obj, HLS) |
|
36
|
|
|
self.assertEqual(hls_obj.media.input, self.src_video) |
|
37
|
|
|
|
|
38
|
|
|
hls_obj.representations(Representation(Size(256, 144), Bitrate(102400))) |
|
39
|
|
|
rep_1 = hls_obj.reps[0] |
|
40
|
|
|
self.assertIsInstance(rep_1, Representation) |
|
41
|
|
|
self.assertEqual(str(rep_1.size), '256x144') |
|
42
|
|
|
self.assertEqual(rep_1.bitrate.video, '100k') |
|
43
|
|
|
|
|
44
|
|
|
hls_obj.auto_generate_representations() |
|
45
|
|
|
reps = list(hls_obj.reps) |
|
46
|
|
|
self.assertEqual(len(reps), 3) |
|
47
|
|
|
|
|
48
|
|
|
for rep_ in reps: |
|
49
|
|
|
self.assertIsInstance(rep_, Representation) |
|
50
|
|
|
self.assertEqual(str(reps[0].size), '480x270') |
|
51
|
|
|
self.assertEqual(reps[0].bitrate.video, '176k') |
|
52
|
|
|
self.assertEqual(str(reps[1].size), '426x240') |
|
53
|
|
|
self.assertEqual(reps[1].bitrate.video, '88k') |
|
54
|
|
|
self.assertEqual(str(reps[2].size), '256x144') |
|
55
|
|
|
self.assertEqual(reps[2].bitrate.video, '71k') |
|
56
|
|
|
|
|
57
|
|
|
hls_obj.output(os.path.join(self.src_dir, 'hls', 'test.m3u8')) |
|
58
|
|
|
with open(os.path.join(self.src_dir, 'fixture_test.m3u8')) as test_m3u8: |
|
59
|
|
|
expected_m3u8 = test_m3u8.read() |
|
60
|
|
|
with open(os.path.join(self.src_dir, 'hls', 'test.m3u8')) as test_m3u8: |
|
61
|
|
|
actual_m3u8 = test_m3u8.read() |
|
62
|
|
|
self.assertEqual(actual_m3u8, expected_m3u8) |
|
63
|
|
|
with open(os.path.join(self.src_dir, 'hls', 'test_270p.m3u8')) as test_m3u8: |
|
64
|
|
|
actual_270_m3u8 = test_m3u8.readlines() |
|
65
|
|
|
self.assertEqual(actual_270_m3u8[0].replace('\n', ''), '#EXTM3U') |
|
66
|
|
|
self.assertEqual(actual_270_m3u8[1].replace('\n', ''), '#EXT-X-VERSION:3') |
|
67
|
|
|
self.assertEqual(actual_270_m3u8[2].replace('\n', ''), '#EXT-X-ALLOW-CACHE:YES') |
|
68
|
|
|
|
|
69
|
|
|
def test_encrypted_hls(self): |
|
70
|
|
|
video = ffmpeg_streaming.input(self.src_video) |
|
71
|
|
|
hls_obj = video.hls(Formats.h264()) |
|
72
|
|
|
hls_obj.encryption(os.path.join(self.src_dir, 'enc.key'), 'https://www.aminyazdanpanah.com/enc.key') |
|
73
|
|
|
|
|
74
|
|
|
self.assertIsNotNone(hls_obj.options.get('hls_key_info_file', None)) |
|
75
|
|
|
|
|
76
|
|
|
with open(hls_obj.options.get('hls_key_info_file', None)) as key_info: |
|
77
|
|
|
key_info = key_info.readlines() |
|
78
|
|
|
self.assertEqual(key_info[0].replace('\n', ''), 'https://www.aminyazdanpanah.com/enc.key') |
|
79
|
|
|
self.assertEqual(key_info[1].replace('\n', ''), os.path.join(self.src_dir, 'enc.key')) |
|
80
|
|
|
|
|
81
|
|
|
hls_obj.auto_generate_representations() |
|
82
|
|
|
|
|
83
|
|
|
hls_obj.output(os.path.join(self.src_dir, 'encrypted_hls', 'test.m3u8'), stderr=False) |
|
84
|
|
|
with open(os.path.join(self.src_dir, 'fixture_test.m3u8')) as test_m3u8: |
|
85
|
|
|
expected_m3u8 = test_m3u8.read() |
|
86
|
|
|
with open(os.path.join(self.src_dir, 'encrypted_hls', 'test.m3u8')) as test_m3u8: |
|
87
|
|
|
actual_encrypted_m3u8 = test_m3u8.read() |
|
88
|
|
|
self.assertEqual(actual_encrypted_m3u8, expected_m3u8) |
|
89
|
|
|
|
|
90
|
|
|
def test_dash(self): |
|
91
|
|
|
video = ffmpeg_streaming.input(self.src_video) |
|
92
|
|
|
dash_obj = video.dash(Formats.hevc()) |
|
93
|
|
|
self.assertIsInstance(dash_obj, DASH) |
|
94
|
|
|
self.assertEqual(dash_obj.media.input, self.src_video) |
|
95
|
|
|
|
|
96
|
|
|
dash_obj.auto_generate_representations() |
|
97
|
|
|
|
|
98
|
|
|
dash_obj.output(os.path.join(self.src_dir, 'dash', 'test.mpd'), stderr=False) |
|
99
|
|
|
with open(os.path.join(self.src_dir, 'dash', 'test.mpd')) as test_mpd: |
|
100
|
|
|
actual_mpd = test_mpd.readlines() |
|
101
|
|
|
self.assertEqual(actual_mpd[0].replace('\n', ''), '<?xml version="1.0" encoding="utf-8"?>') |
|
102
|
|
|
self.assertEqual(actual_mpd[1].replace('\n', ''), '<MPD xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"') |
|
103
|
|
|
|
|
104
|
|
|
|
|
105
|
|
|
if __name__ == '__main__': |
|
106
|
|
|
unittest.main() |
|
107
|
|
|
|
aminyazdanpanah /
python-ffmpeg-video-streaming
