File: test_rtpplay.py

package info (click to toggle)
streamlink 8.1.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 5,568 kB
  • sloc: python: 51,299; sh: 184; makefile: 152
file content (99 lines) | stat: -rw-r--r-- 3,047 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
from __future__ import annotations

from typing import TYPE_CHECKING

import pytest

from streamlink.plugins.rtpplay import RTPPlay
from tests.plugins import PluginCanHandleUrl
from tests.resources import text


if TYPE_CHECKING:
    from requests_mock import Mocker

    from streamlink import Streamlink


class TestPluginCanHandleUrlRTPPlay(PluginCanHandleUrl):
    __plugin__ = RTPPlay

    should_match = [
        "http://www.rtp.pt/play/",
        "https://www.rtp.pt/play/",
        "https://www.rtp.pt/play/direto/rtp1",
        "https://www.rtp.pt/play/direto/rtpmadeira",
    ]

    should_not_match = [
        "https://www.rtp.pt/programa/",
        "http://www.rtp.pt/programa/",
        "https://media.rtp.pt/",
        "http://media.rtp.pt/",
    ]


class TestRTPPlay:
    # all invalid HLS URLs at the beginning need to be ignored ("https://invalid")
    _content_pre = """
        /*  var player1 = new RTPPlayer({
                file: {hls : atob( decodeURIComponent(["aHR0c", "HM6Ly", "9pbnZ", "hbGlk"].join("") ) ), dash : foo() } }); */
        var f = {hls : atob( decodeURIComponent(["aHR0c", "HM6Ly", "9pbnZ", "hbGlk"].join("") ) ), dash: foo() };
    """
    # invalid resources sometimes have an empty string as HLS URL
    _content_invalid = """
        var f = {hls : ""};
    """
    # the actual HLS URL always comes last ("https://valid")
    _content_valid = """
        var f = {hls : decodeURIComponent(["https%3","A%2F%2F", "valid"].join("") ), dash: foo() };
    """
    _content_valid_b64 = """
        var f = {hls : atob( decodeURIComponent(["aHR0c", "HM6Ly", "92YWx", "pZA=="].join("") ) ), dash: foo() };
    """

    @pytest.fixture(scope="class")
    def playlist(self):
        with text("hls/test_master.m3u8") as fd:
            return fd.read()

    @pytest.fixture()
    def streams(self, request: pytest.FixtureRequest, requests_mock: Mocker, session: Streamlink, playlist: str):
        response = getattr(request, "param", "")

        requests_mock.get("https://www.rtp.pt/play/id/title", text=response)
        requests_mock.get("https://valid", text=playlist)
        requests_mock.get("https://invalid", exc=AssertionError)

        plugin = RTPPlay(session, "https://www.rtp.pt/play/id/title")

        return plugin._get_streams()

    @pytest.mark.parametrize(
        ("streams", "expected"),
        [
            pytest.param(
                "",
                False,
                id="empty",
            ),
            pytest.param(
                _content_pre + _content_invalid,
                False,
                id="invalid",
            ),
            pytest.param(
                _content_pre + _content_valid,
                True,
                id="valid",
            ),
            pytest.param(
                _content_pre + _content_valid_b64,
                True,
                id="valid-b64",
            ),
        ],
        indirect=["streams"],
    )
    def test_streams(self, streams, expected):
        assert (streams is not None) is expected