from __future__ import annotations import json import unittest from contextlib import nullcontext from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING from unittest.mock import MagicMock, Mock, call, patch from urllib.parse import parse_qsl, urlparse import pytest from streamlink.exceptions import NoStreamsError, PluginError from streamlink.options import Options from streamlink.plugin.api import useragents from streamlink.plugins.twitch import Twitch, TwitchAPI, TwitchHLSStream, TwitchHLSStreamReader, TwitchHLSStreamWriter from tests.mixins.stream_hls import EventedHLSStreamWriter, Playlist, Segment as _Segment, Tag, TestMixinStreamHLS from tests.plugins import PluginCanHandleUrl from tests.resources import text if TYPE_CHECKING: import requests_mock as rm from streamlink import Streamlink class TestPluginCanHandleUrlTwitch(PluginCanHandleUrl): __plugin__ = Twitch should_match_groups = [ ( ("live", "https://www.twitch.tv/CHANNELNAME"), { "channel": "CHANNELNAME", }, ), ( ("live", "https://www.twitch.tv/CHANNELNAME?"), { "channel": "CHANNELNAME", }, ), ( ("live", "https://www.twitch.tv/CHANNELNAME/"), { "channel": "CHANNELNAME", }, ), ( ("live", "https://www.twitch.tv/CHANNELNAME/?"), { "channel": "CHANNELNAME", }, ), ( ("vod", "https://www.twitch.tv/videos/1963401646"), { "video_id": "1963401646", }, ), ( ("vod", "https://www.twitch.tv/dota2ti/v/1963401646"), { "video_id": "1963401646", }, ), ( ("vod", "https://www.twitch.tv/dota2ti/video/1963401646"), { "video_id": "1963401646", }, ), ( ("vod", "https://www.twitch.tv/videos/1963401646?t=1h23m45s"), { "video_id": "1963401646", }, ), ( ("clip", "https://clips.twitch.tv/GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4"), { "clip_id": "GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4", }, ), ( ("clip", "https://www.twitch.tv/clip/GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4"), { "clip_id": "GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4", }, ), ( ("clip", "https://www.twitch.tv/lirik/clip/GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4"), { "clip_id": "GoodEndearingPassionfruitPMSTwin-QfRLYDPKlscgqt-4", }, ), ( ("clip", "https://twitch.tv/papaplatte/clip/SmellyDeadMomBloodTrail-WWr5gMxd0pe0BAge"), { "clip_id": "SmellyDeadMomBloodTrail-WWr5gMxd0pe0BAge", }, ), ( ("player", "https://player.twitch.tv/?parent=twitch.tv&channel=CHANNELNAME"), {}, ), ( ("player", "https://player.twitch.tv/?parent=twitch.tv&video=1963401646"), {}, ), ( ("player", "https://player.twitch.tv/?parent=twitch.tv&video=1963401646&t=1h23m45s"), {}, ), ] should_not_match = [ "https://www.twitch.tv", "https://www.twitch.tv/", "https://www.twitch.tv/videos/", "https://www.twitch.tv/dota2ti/v", "https://www.twitch.tv/dota2ti/video/", "https://clips.twitch.tv/", "https://www.twitch.tv/clip/", "https://www.twitch.tv/lirik/clip/", "https://player.twitch.tv/", "https://player.twitch.tv/?", ] DATETIME_BASE = datetime(2000, 1, 1, 0, 0, 0, 0, timezone.utc) DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" class TagDateRangeAd(Tag): def __init__(self, start=DATETIME_BASE, duration=1, attrid="stitched-ad-1234", classname="twitch-stitched-ad", custom=None): attrs = { "ID": self.val_quoted_string(attrid), "CLASS": self.val_quoted_string(classname), "START-DATE": self.val_quoted_string(start.strftime(DATETIME_FORMAT)), "DURATION": duration, } if custom is not None: attrs.update(**{key: self.val_quoted_string(value) for (key, value) in custom.items()}) super().__init__("EXT-X-DATERANGE", attrs) class Segment(_Segment): def __init__(self, num, title="live", *args, **kwargs): super().__init__(num, title, *args, **kwargs) self.date = DATETIME_BASE + timedelta(seconds=num) def build(self, namespace): return "#EXT-X-PROGRAM-DATE-TIME:{0}\n{1}".format( self.date.strftime(DATETIME_FORMAT), super().build(namespace), ) class SegmentPrefetch(Segment): def build(self, namespace): return "#EXT-X-TWITCH-PREFETCH:{0}".format(self.url(namespace)) class _TwitchHLSStreamWriter(EventedHLSStreamWriter, TwitchHLSStreamWriter): pass class _TwitchHLSStreamReader(TwitchHLSStreamReader): __writer__ = _TwitchHLSStreamWriter class _TwitchHLSStream(TwitchHLSStream): __reader__ = _TwitchHLSStreamReader def test_stream_weight(requests_mock: rm.Mocker, session: Streamlink): plugin = Twitch(session, "http://twitch.tv/foo") with text("hls/test_master_twitch_vod.m3u8") as fh: playlist = fh.read() requests_mock.request(method="GET", url="http://mocked/master.m3u8", text=playlist) streams = TwitchHLSStream.parse_variant_playlist(session, "http://mocked/master.m3u8") with patch.object(plugin, "_get_streams", return_value=streams): data = plugin.streams() assert list(data.keys()) == ["audio", "160p30", "360p30", "480p30", "720p30", "720p60", "source", "worst", "best"] assert data["best"] is data["source"] assert data["worst"] is data["160p30"] @patch("streamlink.stream.hls.HLSStreamWorker.wait", MagicMock(return_value=True)) class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase): __stream__ = _TwitchHLSStream stream: TwitchHLSStream def get_session(self, *args, **kwargs): session = super().get_session(*args, **kwargs) session.set_option("hls-live-edge", 4) return session def test_hls_daterange_unknown(self): daterange = TagDateRangeAd( start=DATETIME_BASE, duration=1, attrid="foo", classname="bar", custom=None, ) segments = self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(2) data = self.await_read(read_all=True) assert data == self.content(segments), "Doesn't filter out segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" def test_hls_daterange_by_class(self): daterange = TagDateRangeAd( start=DATETIME_BASE, duration=1, attrid="foo", classname="twitch-stitched-ad", custom=None, ) segments = self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(2) data = self.await_read(read_all=True) assert data == segments[1].content, "Filters out ad segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" def test_hls_daterange_by_id(self): daterange = TagDateRangeAd( start=DATETIME_BASE, duration=1, attrid="stitched-ad-1234", classname="/", custom=None, ) segments = self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(2) data = self.await_read(read_all=True) assert data == segments[1].content, "Filters out ad segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" @patch("streamlink.plugins.twitch.log") def test_hls_has_preroll(self, mock_log): daterange = TagDateRangeAd( duration=4, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "PREROLL"}, ) segments = self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1)]), Playlist(2, [daterange, Segment(2), Segment(3)]), Playlist(4, [Segment(4), Segment(5)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(6) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: s.num >= 4), "Filters out preroll ad segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" assert self.thread.reader.worker.duration == 2.0, "Ad segments don't affect the output duration" assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Waiting for pre-roll ads to finish, be patient"), call("Detected advertisement break of 4 seconds"), ] @patch("streamlink.plugins.twitch.log") def test_hls_has_midroll(self, mock_log): daterange = TagDateRangeAd( start=DATETIME_BASE + timedelta(seconds=2), duration=2, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "MIDROLL", "X-TV-TWITCH-AD-COMMERCIAL-ID": "123"}, ) segments = self.subject( [ Playlist(0, [Segment(0), Segment(1)]), Playlist(2, [daterange, Segment(2), Segment(3)]), Playlist(4, [Segment(4), Segment(5)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(6) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: s.num != 2 and s.num != 3), "Filters out mid-stream ad segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" assert self.thread.reader.worker.duration == 4.0, "Ad segments don't affect the output duration" assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Detected advertisement break of 2 seconds"), ] @patch("streamlink.plugins.twitch.log") def test_hls_has_preroll_and_midroll(self, mock_log): ads1a = TagDateRangeAd( start=DATETIME_BASE, duration=2, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "PREROLL"}, ) ads1b = TagDateRangeAd( start=DATETIME_BASE, duration=1, ) ads2 = TagDateRangeAd( start=DATETIME_BASE + timedelta(seconds=4), duration=4, custom={ "X-TV-TWITCH-AD-ROLL-TYPE": "MIDROLL", "X-TV-TWITCH-AD-COMMERCIAL-ID": "123", }, ) ads3 = TagDateRangeAd( start=DATETIME_BASE + timedelta(seconds=8), duration=1, custom={ "X-TV-TWITCH-AD-ROLL-TYPE": "MIDROLL", "X-TV-TWITCH-AD-COMMERCIAL-ID": "456", "X-TV-TWITCH-AD-POD-FILLED-DURATION": ".9", }, ) segments = self.subject( [ Playlist(0, [ads1a, ads1b, Segment(0)]), Playlist(1, [ads1a, ads1b, Segment(1)]), Playlist(2, [Segment(2), Segment(3)]), Playlist(4, [ads2, Segment(4), Segment(5)]), Playlist(6, [ads2, Segment(6), Segment(7)]), Playlist(8, [ads3, Segment(8), Segment(9)], end=True), ], streamoptions={"low_latency": False}, ) self.await_write(10) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: s.num not in (0, 1, 4, 5, 6, 7, 8)), "Filters out all ad segments" assert all(self.called(s) for s in segments.values()), "Downloads all segments" assert self.thread.reader.worker.duration == 3.0, "Ad segments don't affect the output duration" assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Waiting for pre-roll ads to finish, be patient"), call("Detected advertisement break of 2 seconds"), call("Detected advertisement break of 4 seconds"), call("Detected advertisement break of 1 second"), ] @patch("streamlink.plugins.twitch.log") def test_hls_low_latency_has_prefetch(self, mock_log): segments = self.subject( [ Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3), SegmentPrefetch(4), SegmentPrefetch(5)]), Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True), ], streamoptions={"low_latency": True}, ) assert self.session.options.get("hls-live-edge") == 2 assert self.session.options.get("hls-segment-stream-data") self.await_write(6) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: s.num >= 4), "Skips first four segments due to reduced live-edge" assert not any(self.called(s) for s in segments.values() if s.num < 4), "Doesn't download old segments" assert all(self.called(s) for s in segments.values() if s.num >= 4), "Downloads all remaining segments" assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Low latency streaming (HLS live edge: 2)"), ] @patch("streamlink.plugins.twitch.log") def test_hls_no_low_latency_has_prefetch(self, mock_log): segments = self.subject( [ Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3), SegmentPrefetch(4), SegmentPrefetch(5)]), Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True), ], streamoptions={"low_latency": False}, ) assert self.session.options.get("hls-live-edge") == 4 assert not self.session.options.get("hls-segment-stream-data") self.await_write(8) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: s.num < 8), "Ignores prefetch segments" assert all(self.called(s) for s in segments.values() if s.num <= 7), "Ignores prefetch segments" assert not any(self.called(s) for s in segments.values() if s.num > 7), "Ignores prefetch segments" assert mock_log.info.mock_calls == [ call("Will skip ad segments"), ] assert self.thread.reader.worker._reload_time == 3.0 @patch("streamlink.plugins.twitch.log") def test_hls_low_latency_no_prefetch(self, mock_log): self.subject( [ Playlist(0, [Segment(0), Segment(1), Segment(2), Segment(3)]), Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7)], end=True), ], streamoptions={"low_latency": True}, ) assert self.stream.low_latency self.await_write(6) self.await_read(read_all=True) assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Low latency streaming (HLS live edge: 2)"), call("This is not a low latency stream"), ] @patch("streamlink.plugins.twitch.log") def test_hls_low_latency_has_prefetch_has_preroll(self, mock_log): daterange = TagDateRangeAd( duration=4, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "PREROLL"}, ) self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1), Segment(2), Segment(3)]), Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(8) self.await_read(read_all=True) assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Low latency streaming (HLS live edge: 2)"), call("Waiting for pre-roll ads to finish, be patient"), call("Detected advertisement break of 4 seconds"), ] @patch("streamlink.plugins.twitch.log") def test_hls_low_latency_has_prefetch_no_preroll_with_prefetch_ads(self, mock_log): # segment 1 has a shorter duration, to mess with the extrapolation of the prefetch start times # segments 3-6 are ads Seg, Pre = Segment, SegmentPrefetch ads = [ Tag("EXT-X-DISCONTINUITY"), TagDateRangeAd( start=DATETIME_BASE + timedelta(seconds=3), duration=4, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "MIDROLL"}, ), ] tls = Tag("EXT-X-TWITCH-LIVE-SEQUENCE", 7) # noinspection PyTypeChecker segments = self.subject( [ # regular stream data with prefetch segments Playlist(0, [Seg(0), Seg(1, duration=0.5), Pre(2), Pre(3)]), # three prefetch segments, one regular (2) and two ads (3 and 4) Playlist(1, [Seg(1, duration=0.5), Pre(2), *ads, Pre(3), Pre(4)]), # all prefetch segments are gone once regular prefetch segments have shifted Playlist(2, [Seg(2, duration=1.5), *ads, Seg(3), Seg(4), Seg(5)]), # still no prefetch segments while ads are playing Playlist(3, [*ads, Seg(3), Seg(4), Seg(5), Seg(6)]), # new prefetch segments on the first regular segment occurrence Playlist(4, [*ads, Seg(4), Seg(5), Seg(6), tls, Seg(7), Pre(8), Pre(9)]), Playlist(5, [*ads, Seg(5), Seg(6), tls, Seg(7), Seg(8), Pre(9), Pre(10)]), Playlist(6, [*ads, Seg(6), tls, Seg(7), Seg(8), Seg(9), Pre(10), Pre(11)]), Playlist(7, [Seg(7), Seg(8), Seg(9), Seg(10), Pre(11), Pre(12)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(11) data = self.await_read(read_all=True) assert data == self.content(segments, cond=lambda s: 2 <= s.num <= 3 or 7 <= s.num) assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Low latency streaming (HLS live edge: 2)"), call("Detected advertisement break of 4 seconds"), ] @patch("streamlink.plugins.twitch.log") def test_hls_low_latency_no_prefetch_has_preroll(self, mock_log): daterange = TagDateRangeAd( duration=4, custom={"X-TV-TWITCH-AD-ROLL-TYPE": "PREROLL"}, ) self.subject( [ Playlist(0, [daterange, Segment(0), Segment(1), Segment(2), Segment(3)]), Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(6) self.await_read(read_all=True) assert mock_log.info.mock_calls == [ call("Will skip ad segments"), call("Low latency streaming (HLS live edge: 2)"), call("Waiting for pre-roll ads to finish, be patient"), call("Detected advertisement break of 4 seconds"), call("This is not a low latency stream"), ] def test_hls_low_latency_no_ads_reload_time(self): Seg, SegPre = Segment, SegmentPrefetch self.subject( [ Playlist(0, [Seg(0, duration=5), Seg(1, duration=7), Seg(2, duration=11), SegPre(3)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(4) self.await_read(read_all=True) assert self.thread.reader.worker._reload_time == pytest.approx(23 / 3) @patch("streamlink.stream.hls.hls.log") def test_hls_prefetch_after_discontinuity(self, mock_log): segments = self.subject( [ Playlist(0, [Segment(0), Segment(1)]), Playlist(2, [Segment(2), Segment(3), Tag("EXT-X-DISCONTINUITY"), SegmentPrefetch(4), SegmentPrefetch(5)]), Playlist(6, [Segment(6), Segment(7)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(8) assert self.await_read(read_all=True) == self.content(segments, cond=lambda seg: seg.num not in (4, 5)) assert mock_log.warning.mock_calls == [ call("Encountered a stream discontinuity. This is unsupported and will result in incoherent output data."), ] @patch("streamlink.stream.hls.hls.log") def test_hls_ignored_discontinuity(self, mock_log): Seg, Pre = Segment, SegmentPrefetch discontinuity = Tag("EXT-X-DISCONTINUITY") tls = Tag("EXT-X-TWITCH-LIVE-SEQUENCE", 1234) # value is irrelevant segments = self.subject( [ Playlist(0, [Seg(0), discontinuity, Seg(1)]), Playlist(2, [Seg(2), Seg(3), discontinuity, Seg(4), Seg(5)]), Playlist(6, [Seg(6), Seg(7), discontinuity, tls, Pre(8), Pre(9)]), Playlist(10, [Seg(10), Seg(11), discontinuity, tls, Pre(12), discontinuity, tls, Pre(13)], end=True), ], streamoptions={"low_latency": True}, ) self.await_write(14) assert self.await_read(read_all=True) == self.content(segments) assert mock_log.warning.mock_calls == [] class TestUsherService: @pytest.fixture(autouse=True) def caplog(self, caplog: pytest.LogCaptureFixture): caplog.set_level(1, "streamlink.plugins.twitch") return caplog @pytest.fixture() def plugin(self, request: pytest.FixtureRequest, session: Streamlink): return Twitch( session, "https://twitch.tv/twitch", options=Options(getattr(request, "param", {})), ) @pytest.fixture() def endpoint(self, request: pytest.FixtureRequest, caplog: pytest.LogCaptureFixture, plugin: Twitch): param = getattr(request, "param", {}) service = param.get("service", "channel") args = param.get("args", ("twitch",)) token = { "expires": 9876543210, "channel": "twitch", "channel_id": 123, "user_id": 456, "user_ip": "127.0.0.1", "adblock": False, "geoblock_reason": "", "hide_ads": False, "server_ads": True, "show_ads": True, } return getattr(plugin.usher, service)(*args, token=json.dumps(token), sig="tokensignature") @pytest.mark.parametrize( ("endpoint", "expected_path", "logs"), [ pytest.param( {"service": "channel", "args": ("TWITCH",)}, "/api/channel/hls/twitch.m3u8", [ ( "streamlink.plugins.twitch", "debug", "{'adblock': False, 'geoblock_reason': '', 'hide_ads': False, 'server_ads': True, 'show_ads': True}", ), ], id="channel", ), pytest.param( {"service": "video", "args": ("1234567890",)}, "/vod/1234567890", [], id="video", ), ], indirect=["endpoint"], ) def test_service(self, caplog: pytest.LogCaptureFixture, endpoint: str, expected_path: str, logs: list): url = urlparse(endpoint) assert url.path == expected_path qs = dict(parse_qsl(url.query)) assert qs.get("token") assert qs.get("sig") assert [(r.name, r.levelname, r.message) for r in caplog.get_records(when="setup")] == logs @pytest.mark.parametrize( "endpoint", [ pytest.param({"service": "channel"}, id="channel"), pytest.param({"service": "video"}, id="video"), ], indirect=True, ) @pytest.mark.parametrize( ("plugin", "expected"), [ pytest.param({}, "h264", id="unset"), pytest.param({"supported_codecs": []}, "h264", id="empty"), pytest.param({"supported_codecs": ["h264"]}, "h264", id="h264"), pytest.param({"supported_codecs": ["av1", "h264"]}, "av1,h264", id="av1,h264"), pytest.param({"supported_codecs": ["av1", "h264", "h265"]}, "av1,h264,h265", id="av1,h264,h265"), ], indirect=["plugin"], ) def test_supported_codecs(self, plugin: Twitch, endpoint: str, expected: str): qs = dict(parse_qsl(urlparse(endpoint).query)) assert qs.get("supported_codecs") == expected class TestTwitchAPIAccessToken: @pytest.fixture(autouse=True) def _client_integrity_token(self, monkeypatch: pytest.MonkeyPatch): mock_client_integrity_token = Mock(return_value=("device-id", "client-integrity-token")) monkeypatch.setattr(Twitch, "_client_integrity_token", mock_client_integrity_token) @pytest.fixture() def plugin(self, request: pytest.FixtureRequest, session: Streamlink): options = Options(getattr(request, "param", {})) return Twitch(session, "https://twitch.tv/channelname", options) @pytest.fixture() def mock(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker): mock = requests_mock.post("https://gql.twitch.tv/gql", **getattr(request, "param", {"json": {}})) yield mock assert mock.call_count > 0 payload = mock.last_request.json() # type: ignore[union-attr] assert tuple(sorted(payload.keys())) == ("extensions", "operationName", "variables") assert payload.get("operationName") == "PlaybackAccessToken" assert payload.get("extensions") == { "persistedQuery": { "sha256Hash": "ed230aa1e33e07eebb8928504583da78a5173989fadfb1ac94be06a04f3cdbe9", "version": 1, }, } @pytest.fixture() def _assert_live(self, mock: rm.Mocker): yield assert mock.last_request.json().get("variables") == { # type: ignore[union-attr] "isLive": True, "isVod": False, "login": "channelname", "vodID": "", "playerType": "embed", "platform": "site", } @pytest.fixture() def _assert_vod(self, mock: rm.Mocker): yield assert mock.last_request.json().get("variables") == { # type: ignore[union-attr] "isLive": False, "isVod": True, "login": "", "vodID": "vodid", "playerType": "embed", "platform": "site", } @pytest.mark.parametrize( ("plugin", "exp_headers", "exp_variables"), [ ( {}, {"Client-ID": TwitchAPI.CLIENT_ID}, { "isLive": True, "isVod": False, "login": "channelname", "vodID": "", "playerType": "embed", "platform": "site", }, ), ( { "api-header": [ ("Authorization", "invalid data"), ("Authorization", "OAuth 0123456789abcdefghijklmnopqrst"), ], "access-token-param": [ ("specialVariable", "specialValue"), ("playerType", "frontpage"), ("platform", "other"), ], }, { "Client-ID": TwitchAPI.CLIENT_ID, "Authorization": "OAuth 0123456789abcdefghijklmnopqrst", }, { "isLive": True, "isVod": False, "login": "channelname", "vodID": "", "playerType": "frontpage", "platform": "other", "specialVariable": "specialValue", }, ), ], indirect=["plugin"], ) def test_plugin_options(self, plugin: Twitch, mock: rm.Mocker, exp_headers: dict, exp_variables: dict): with pytest.raises(PluginError): plugin._access_token(True, "channelname") requestheaders = dict(mock.last_request._request.headers) # type: ignore[union-attr] for header in plugin.session.http.headers.keys(): del requestheaders[header] del requestheaders["Content-Type"] del requestheaders["Content-Length"] assert requestheaders == exp_headers assert mock.last_request.json().get("variables") == exp_variables # type: ignore[union-attr] @pytest.mark.parametrize( ("session", "mock"), [ pytest.param( {}, {"json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}}, id="no-custom-user-agent", ), pytest.param( {"http-headers": {"User-Agent": "foo"}}, {"json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}}, id="custom-user-agent", ), ], indirect=True, ) def test_user_agent(self, plugin: Twitch, mock: rm.Mocker): plugin._access_token(True, "channelname") assert len(mock.request_history) > 0 assert mock.request_history[0]._request.headers["User-Agent"] == useragents.DEFAULT @pytest.mark.usefixtures("_assert_live") @pytest.mark.parametrize( ("plugin", "mock"), [ pytest.param( { "force-client-integrity": False, }, { "json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, id="no-force-client-integrity", ), pytest.param( { "force-client-integrity": True, }, { "json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, id="force-client-integrity", ), ], indirect=True, ) def test_live_success(self, plugin: Twitch, mock: rm.Mocker): data = plugin._access_token(True, "channelname") assert data == ("sig", '{"channel":"foo"}', []) assert len(mock.request_history) == 1 @pytest.mark.usefixtures("_assert_live") @pytest.mark.parametrize( "mock", [ { "json": {"data": {"streamPlaybackAccessToken": None}}, }, ], indirect=True, ) def test_live_failure(self, plugin: Twitch, mock: rm.Mocker): with pytest.raises(NoStreamsError): plugin._access_token(True, "channelname") assert len(mock.request_history) == 1, "Only gets the access token once when the channel is offline" @pytest.mark.usefixtures("_assert_vod") @pytest.mark.parametrize( "mock", [ { "json": {"data": {"videoPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, ], indirect=True, ) def test_vod_success(self, plugin: Twitch, mock: rm.Mocker): data = plugin._access_token(False, "vodid") assert data == ("sig", '{"channel":"foo"}', []) @pytest.mark.usefixtures("_assert_vod") @pytest.mark.parametrize( "mock", [ { "json": {"data": {"videoPlaybackAccessToken": None}}, }, ], indirect=True, ) def test_vod_failure(self, plugin: Twitch, mock: rm.Mocker): with pytest.raises(NoStreamsError): plugin._access_token(False, "vodid") assert len(mock.request_history) == 1, "Only gets the access token once when the VOD doesn't exist" @pytest.mark.usefixtures("_assert_live") @pytest.mark.parametrize( ("plugin", "mock"), [ ( { "api-header": [("Authorization", "OAuth invalid-token")], }, { "status_code": 401, "json": {"error": "Unauthorized", "status": 401, "message": 'The "Authorization" token is invalid.'}, }, ), ], indirect=True, ) def test_auth_failure(self, plugin: Twitch, mock: rm.Mocker): with pytest.raises(PluginError, match=r'^Unauthorized: The "Authorization" token is invalid\.$'): plugin._access_token(True, "channelname") assert len(mock.request_history) == 2, "Always tries again on error, with integrity-token on second attempt" headers: dict = mock.request_history[0]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert "Device-Id" not in headers assert "Client-Integrity" not in headers headers = mock.request_history[1]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert headers["Device-Id"] == "device-id" assert headers["Client-Integrity"] == "client-integrity-token" @pytest.mark.usefixtures("_assert_live") @pytest.mark.parametrize( ("plugin", "mock"), [ ( { "force-client-integrity": False, "api-header": [("Authorization", "OAuth invalid-token")], }, { "response_list": [ { "status_code": 401, "json": {"errors": [{"message": "failed integrity check"}]}, }, { "json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, ], }, ), ], indirect=True, ) def test_integrity_check_not_forced(self, plugin: Twitch, mock: rm.Mocker): data = plugin._access_token(True, "channelname") assert data == ("sig", '{"channel":"foo"}', []) assert len(mock.request_history) == 2, "Always tries again on error, with integrity-token on second attempt" headers: dict = mock.request_history[0]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert "Device-Id" not in headers assert "Client-Integrity" not in headers headers = mock.request_history[1]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert headers["Device-Id"] == "device-id" assert headers["Client-Integrity"] == "client-integrity-token" @pytest.mark.usefixtures("_assert_live") @pytest.mark.parametrize( ("plugin", "mock"), [ ( { "force-client-integrity": True, "api-header": [("Authorization", "OAuth invalid-token")], }, { "response_list": [ { "json": {"data": {"streamPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, ], }, ), ], indirect=True, ) def test_integrity_check_forced(self, plugin: Twitch, mock: rm.Mocker): data = plugin._access_token(True, "channelname") assert data == ("sig", '{"channel":"foo"}', []) assert len(mock.request_history) == 1 headers: dict = mock.request_history[0]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert headers["Device-Id"] == "device-id" assert headers["Client-Integrity"] == "client-integrity-token" @pytest.mark.usefixtures("_assert_vod") @pytest.mark.parametrize( ("plugin", "mock"), [ ( { "force-client-integrity": True, "api-header": [("Authorization", "OAuth invalid-token")], }, { "response_list": [ { "json": {"data": {"videoPlaybackAccessToken": {"value": '{"channel":"foo"}', "signature": "sig"}}}, }, ], }, ), ], indirect=True, ) def test_integrity_check_vod(self, plugin: Twitch, mock: rm.Mocker): data = plugin._access_token(False, "vodid") assert data == ("sig", '{"channel":"foo"}', []) assert len(mock.request_history) == 1 headers: dict = mock.request_history[0]._request.headers assert headers["Authorization"] == "OAuth invalid-token" assert "Device-Id" not in headers assert "Client-Integrity" not in headers class TestTwitchHLSMultivariantResponse: @pytest.fixture() def plugin(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker, session: Streamlink): requests_mock.get("mock://multivariant", **getattr(request, "param", {})) return Twitch(session, "https://twitch.tv/channelname") @pytest.mark.parametrize( ("plugin", "streamid", "raises", "streams", "log"), [ pytest.param( {"text": "#EXTM3U\n"}, "123", nullcontext(), {}, [], id="success", ), pytest.param( {"text": "Not an HLS playlist"}, "123", pytest.raises(PluginError), {}, [], id="invalid HLS playlist", ), pytest.param( { "status_code": 404, "json": [ { "url": "mock://multivariant", "error": "twirp error not_found: transcode does not exist", "error_code": "transcode_does_not_exist", "type": "error", }, ], }, None, nullcontext(), None, [], id="offline", ), pytest.param( { "status_code": 403, "json": [ { "url": "mock://multivariant", "error": "Content Restricted In Region", "error_code": "content_geoblocked", "type": "error", }, ], }, "123", nullcontext(), None, [("streamlink.plugins.twitch", "error", "Content Restricted In Region")], id="geo restriction", ), pytest.param( { "status_code": 404, "text": "Not found", }, "123", nullcontext(), None, [], id="non-json error response", ), ], indirect=["plugin"], ) def test_multivariant_response( self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, plugin: Twitch, streamid: str | None, raises: nullcontext, streams: dict | None, log: list, ): caplog.set_level("error", "streamlink.plugins.twitch") monkeypatch.setattr(plugin, "get_id", Mock(return_value=streamid)) with raises: assert plugin._get_hls_streams("mock://multivariant", []) == streams assert [(record.name, record.levelname, record.message) for record in caplog.records] == log class TestTwitchMetadata: @pytest.fixture() def metadata(self, request: pytest.FixtureRequest, session: Streamlink): url = getattr(request, "param", "") plugin = Twitch(session, url) return plugin.get_id(), plugin.get_author(), plugin.get_category(), plugin.get_title() @pytest.fixture() def mock_request_channel(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker): data = getattr(request, "param", True) return requests_mock.post( "https://gql.twitch.tv/gql", json=[ { "data": { "userOrError": {"userDoesNotExist": "error"} if not data else { "displayName": "channel name", }, }, }, { "data": { "user": None if not data else { "lastBroadcast": { "title": "channel status", }, "stream": { "id": "stream id", "game": { "name": "channel game", }, }, }, }, }, ], ) @pytest.fixture() def mock_request_video(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker): data = getattr(request, "param", True) return requests_mock.post( "https://gql.twitch.tv/gql", json={ "data": { "video": None if not data else { "id": "video id", "title": "video title", "game": { "displayName": "video game", }, "owner": { "displayName": "channel name", }, }, }, }, ) @pytest.fixture() def mock_request_clip(self, request: pytest.FixtureRequest, requests_mock: rm.Mocker): data = getattr(request, "param", True) return requests_mock.post( "https://gql.twitch.tv/gql", json={ "data": { "clip": None if not data else { "id": "clip id", "broadcaster": { "displayName": "channel name", }, "game": { "name": "game name", }, "title": "clip title", }, }, }, ) @pytest.mark.parametrize(("mock_request_channel", "metadata"), [(True, "https://twitch.tv/foo")], indirect=True) def test_metadata_channel(self, mock_request_channel, metadata): assert metadata == ("stream id", "channel name", "channel game", "channel status") assert mock_request_channel.call_count == 1 assert mock_request_channel.request_history[0].json() == [ { "operationName": "ChannelShell", "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "fea4573a7bf2644f5b3f2cbbdcbee0d17312e48d2e55f080589d053aad353f11", }, }, "variables": { "login": "foo", }, }, { "operationName": "StreamMetadata", "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "b57f9b910f8cd1a4659d894fe7550ccc81ec9052c01e438b290fd66a040b9b93", }, }, "variables": { "channelLogin": "foo", "includeIsDJ": True, }, }, ] @pytest.mark.parametrize(("mock_request_channel", "metadata"), [(False, "https://twitch.tv/foo")], indirect=True) def test_metadata_channel_no_data(self, mock_request_channel, metadata): assert metadata == (None, None, None, None) assert mock_request_channel.call_count == 1 @pytest.mark.parametrize(("mock_request_video", "metadata"), [(True, "https://twitch.tv/videos/1337")], indirect=True) def test_metadata_video(self, mock_request_video, metadata): assert metadata == ("video id", "channel name", "video game", "video title") assert mock_request_video.call_count == 1 assert mock_request_video.request_history[0].json() == { "operationName": "VideoMetadata", "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "45111672eea2e507f8ba44d101a61862f9c56b11dee09a15634cb75cb9b9084d", }, }, "variables": { "channelLogin": "", "videoID": "1337", }, } @pytest.mark.parametrize(("mock_request_video", "metadata"), [(False, "https://twitch.tv/videos/1337")], indirect=True) def test_metadata_video_no_data(self, mock_request_video, metadata): assert metadata == (None, None, None, None) assert mock_request_video.call_count == 1 @pytest.mark.parametrize(("mock_request_clip", "metadata"), [(True, "https://clips.twitch.tv/foo")], indirect=True) def test_metadata_clip(self, mock_request_clip, metadata): assert metadata == ("clip id", "channel name", "game name", "clip title") assert mock_request_clip.call_count == 1 assert mock_request_clip.request_history[0].json() == { "operationName": "ShareClipRenderStatus", "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "1844261bb449fa51e6167040311da4a7a5f1c34fe71c71a3e0c4f551bc30c698", }, }, "variables": { "slug": "foo", }, } @pytest.mark.parametrize(("mock_request_clip", "metadata"), [(False, "https://clips.twitch.tv/foo")], indirect=True) def test_metadata_clip_no_data(self, mock_request_clip, metadata): assert metadata == (None, None, None, None)