Merge pull request #3223 from bastimeyer/plugins/twitch/fix-metadata-api-response-handling

gravyboat · web-flow · commit 6de76fca8f66 · 2020-10-04T11:46:55.000-07:00
plugins.twitch: fix metadata API response handling
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
@@ -396,6 +396,39 @@ class Twitch(Plugin):
         )
     )
 
+    _schema_metadata_empty = validate.transform(lambda _: (None,) * 3)
+    _schema_metadata_channel = validate.Schema(
+        {
+            "stream": validate.any(
+                validate.all(
+                    {"channel": {
+                        "display_name": validate.text,
+                        "game": validate.text,
+                        "status": validate.text
+                    }},
+                    validate.get("channel"),
+                    validate.transform(lambda ch: (ch["display_name"], ch["status"], ch["game"]))
+                ),
+                validate.all(None, _schema_metadata_empty)
+            )
+        },
+        validate.get("stream")
+    )
+    _schema_metadata_video = validate.Schema(validate.any(
+        validate.all(
+            {
+                "title": validate.text,
+                "game": validate.text,
+                "channel": validate.all(
+                    {"display_name": validate.text},
+                    validate.get("display_name")
+                )
+            },
+            validate.transform(lambda data: (data["channel"], data["title"], data["game"]))
+        ),
+        validate.all({}, _schema_metadata_empty)
+    ))
+
     @classmethod
     def stream_weight(cls, key):
         weight = QUALITY_WEIGHTS.get(key)
@@ -410,17 +443,11 @@ def can_handle_url(cls, url):
 
     def _get_metadata(self):
         if self.video_id:
-            api_res = self.api.videos(self.video_id)
-            self.title = api_res["title"]
-            self.author = api_res["channel"]["display_name"]
-            self.category = api_res["game"]
+            (self.author, self.title, self.category) = self.api.videos(self.video_id, schema=self._schema_metadata_video)
         elif self.clip_name:
             self._get_clips()
         elif self._channel:
-            api_res = self.api.streams(self.channel_id)["stream"]["channel"]
-            self.title = api_res["status"]
-            self.author = api_res["display_name"]
-            self.category = api_res["game"]
+            (self.author, self.title, self.category) = self.api.streams(self.channel_id, schema=self._schema_metadata_channel)
 
     def get_title(self):
         if self.title is None:
diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
@@ -7,6 +7,7 @@
 from tests.mock import MagicMock, call, patch
 
 from streamlink import Streamlink
+from streamlink.plugin import PluginError
 from streamlink.plugins.twitch import Twitch, TwitchHLSStream
 
 
@@ -274,6 +275,81 @@ def test_hls_low_latency_no_prefetch_disable_ads_has_preroll(self, mock_log):
         ])
 
 
+class TestTwitchMetadata(unittest.TestCase):
+    def setUp(self):
+        self.mock = requests_mock.Mocker()
+        self.mock.start()
+
+    def tearDown(self):
+        self.mock.stop()
+
+    def subject(self, url):
+        session = Streamlink()
+        Twitch.bind(session, "tests.plugins.test_twitch")
+        plugin = Twitch(url)
+        return plugin.get_author(), plugin.get_title(), plugin.get_category()
+
+    def subject_channel(self, data=True, failure=False):
+        self.mock.get(
+            "https://api.twitch.tv/kraken/users.json?login=foo",
+            json={"users": [{"_id": 1234}]}
+        )
+        self.mock.get(
+            "https://api.twitch.tv/kraken/streams/1234.json",
+            status_code=200 if not failure else 404,
+            json={"stream": None} if not data else {"stream": {
+                "channel": {
+                    "display_name": "channel name",
+                    "status": "channel status",
+                    "game": "channel game"
+                }
+            }}
+        )
+        return self.subject("https://twitch.tv/foo")
+
+    def subject_video(self, data=True, failure=False):
+        self.mock.get(
+            "https://api.twitch.tv/kraken/videos/1337.json",
+            status_code=200 if not failure else 404,
+            json={} if not data else {
+                "title": "video title",
+                "game": "video game",
+                "channel": {
+                    "display_name": "channel name"
+                }
+            }
+        )
+        return self.subject("https://twitch.tv/videos/1337")
+
+    def test_metadata_channel_exists(self):
+        author, title, category = self.subject_channel()
+        self.assertEqual(author, "channel name")
+        self.assertEqual(title, "channel status")
+        self.assertEqual(category, "channel game")
+
+    def test_metadata_channel_missing(self):
+        metadata = self.subject_channel(data=False)
+        self.assertEqual(metadata, (None, None, None))
+
+    def test_metadata_channel_invalid(self):
+        with self.assertRaises(PluginError):
+            self.subject_channel(failure=True)
+
+    def test_metadata_video_exists(self):
+        author, title, category = self.subject_video()
+        self.assertEqual(author, "channel name")
+        self.assertEqual(title, "video title")
+        self.assertEqual(category, "video game")
+
+    def test_metadata_video_missing(self):
+        metadata = self.subject_video(data=False)
+        self.assertEqual(metadata, (None, None, None))
+
+    def test_metadata_video_invalid(self):
+        with self.assertRaises(PluginError):
+            self.subject_video(failure=True)
+
+
 @patch("streamlink.plugins.twitch.log")
 class TestTwitchReruns(unittest.TestCase):
     log_call = call("Reruns were disabled by command line option")
