File: test_hls.py

package info (click to toggle)
streamlink 8.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 5,564 kB
  • sloc: python: 51,188; sh: 184; makefile: 152
file content (131 lines) | stat: -rw-r--r-- 5,551 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
from __future__ import annotations

from typing import TYPE_CHECKING
from unittest.mock import Mock, call

import pytest

from streamlink.plugin.plugin import LOW_PRIORITY, NO_PRIORITY, NORMAL_PRIORITY
from streamlink.plugins.hls import HLSPlugin
from streamlink.stream.hls import HLSStream
from tests.plugins import PluginCanHandleUrl


if TYPE_CHECKING:
    from streamlink.session import Streamlink


class TestPluginCanHandleUrlHLSPlugin(PluginCanHandleUrl):
    __plugin__ = HLSPlugin

    should_match_groups = [
        # implicit HLS URLs
        ("example.com/foo.m3u8", {"url": "example.com/foo.m3u8"}),
        ("example.com/foo.M3U8", {"url": "example.com/foo.M3U8"}),
        ("example.com/foo.m3u8?bar", {"url": "example.com/foo.m3u8?bar"}),
        ("http://example.com/foo.m3u8", {"url": "http://example.com/foo.m3u8"}),
        ("http://example.com/foo.m3u8?bar", {"url": "http://example.com/foo.m3u8?bar"}),
        ("https://example.com/foo.m3u8", {"url": "https://example.com/foo.m3u8"}),
        ("https://example.com/foo.m3u8?bar", {"url": "https://example.com/foo.m3u8?bar"}),
        ("file://foo.m3u8", {"url": "file://foo.m3u8"}),
        ("file:///foo.m3u8", {"url": "file:///foo.m3u8"}),
        ("file://../foo.m3u8", {"url": "file://../foo.m3u8"}),
        # explicit HLS URLs with protocol prefix
        ("hls://example.com/foo?bar", {"url": "example.com/foo?bar"}),
        ("hls://http://example.com/foo?bar", {"url": "http://example.com/foo?bar"}),
        ("hls://https://example.com/foo?bar", {"url": "https://example.com/foo?bar"}),
        ("hls://file://foo", {"url": "file://foo"}),
        ("hlsvariant://example.com/foo?bar", {"url": "example.com/foo?bar"}),
        ("hlsvariant://http://example.com/foo?bar", {"url": "http://example.com/foo?bar"}),
        ("hlsvariant://https://example.com/foo?bar", {"url": "https://example.com/foo?bar"}),
        ("hlsvariant://file://foo", {"url": "file://foo"}),
        # optional parameters
        ("example.com/foo.m3u8?bar abc=def", {"url": "example.com/foo.m3u8?bar", "params": "abc=def"}),
        ("http://example.com/foo.m3u8?bar abc=def", {"url": "http://example.com/foo.m3u8?bar", "params": "abc=def"}),
        ("https://example.com/foo.m3u8?bar abc=def", {"url": "https://example.com/foo.m3u8?bar", "params": "abc=def"}),
        ("hls://https://example.com/foo?bar abc=def", {"url": "https://example.com/foo?bar", "params": "abc=def"}),
        ("hlsvariant://https://example.com/foo?bar abc=def", {"url": "https://example.com/foo?bar", "params": "abc=def"}),
    ]

    should_not_match = [
        # implicit HLS URLs must have their path end with ".m3u8"
        "example.m3u8",
        "/example.m3u8",
        "example.com/m3u8",
        "example.com/M3U8",
        "example.com/m3u8 abc=def",
        "example.com/foo.m3u8,bar",
        "example.com/foo.m3u8,bar abc=def",
        # missing parameters
        "example.com/foo.m3u8 ",
        "hls://example.com/foo ",
        "hlsvariant://example.com/foo ",
    ]


@pytest.mark.parametrize(
    ("url", "priority"),
    [
        ("http://example.com/foo.m3u8", LOW_PRIORITY),
        ("http://example.com/foo.M3U8", LOW_PRIORITY),
        ("hls://http://example.com/foo.m3u8", NORMAL_PRIORITY),
        ("hls://http://example.com/bar", NORMAL_PRIORITY),
        ("hlsvariant://http://example.com/foo.m3u8", NORMAL_PRIORITY),
        ("hlsvariant://http://example.com/bar", NORMAL_PRIORITY),
        ("http://example.com/bar", NO_PRIORITY),
    ],
)
def test_priority(url, priority):
    assert next((matcher.priority for matcher in HLSPlugin.matchers if matcher.pattern.match(url)), NO_PRIORITY) == priority


@pytest.mark.parametrize(
    ("url", "expected"),
    [
        ("example.com/foo.m3u8", "https://example.com/foo.m3u8"),
        ("http://example.com/foo.m3u8", "http://example.com/foo.m3u8"),
        ("https://example.com/foo.m3u8", "https://example.com/foo.m3u8"),
        ("file://foo.m3u8", "file://foo.m3u8"),
        ("file:///foo.m3u8", "file:///foo.m3u8"),
        ("file://../foo.m3u8", "file://../foo.m3u8"),
        ("hls://example.com/foo", "https://example.com/foo"),
        ("hls://http://example.com/foo", "http://example.com/foo"),
        ("hls://https://example.com/foo", "https://example.com/foo"),
        ("hlsvariant://example.com/foo", "https://example.com/foo"),
        ("hlsvariant://http://example.com/foo", "http://example.com/foo"),
        ("hlsvariant://https://example.com/foo", "https://example.com/foo"),
    ],
)
@pytest.mark.parametrize(
    ("isvariant", "streams"),
    [
        (False, ["live"]),
        (True, ["720p", "1080p"]),
    ],
)
def test_get_streams(
    monkeypatch: pytest.MonkeyPatch,
    session: Streamlink,
    url: str,
    expected: str,
    isvariant: bool,
    streams: list,
):
    monkeypatch.setattr("streamlink.stream.hls.HLSStream.__init__", Mock(return_value=None))

    fakestreams = {}
    if isvariant:
        fakestreams["720p"] = HLSStream(session, url)
        fakestreams["1080p"] = HLSStream(session, url)

    mock_parse_variant_playlist = Mock(return_value=fakestreams)
    monkeypatch.setattr("streamlink.stream.hls.HLSStream.parse_variant_playlist", mock_parse_variant_playlist)

    plugin = HLSPlugin(session, url)
    result = plugin.streams()
    result.pop("worst", None)
    result.pop("best", None)

    assert list(result.keys()) == streams
    assert all(isinstance(s, HLSStream) for s in result.values())
    assert mock_parse_variant_playlist.call_args_list == [call(session, expected)]