Repository: streamlink/streamlink
Pull request #3910 (2021-08-10 10:39:44)

stream.hls: close stream on initial parsing error
- close stream immediately when the first playlist fails to parse
- add tests for all playlist parsing errors in the worker thread

----

Follow up of #3903

gravyboat: Looks good, thanks @bastimeyer.

Patch:
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index 3d1d2271..a94a5579 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -231,7 +231,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
         return hls_playlist.load(text, url)
 
     def reload_playlist(self):
-        if self.closed:
+        if self.closed:  # pragma: no cover
             return
 
         self.reader.buffer.wait_free()
@@ -246,11 +246,10 @@ class HLSStreamWorker(SegmentedStreamWorker):
             raise StreamError(err)
 
         if playlist.is_master:
-            raise StreamError("Attempted to play a variant playlist, use "
-                              "'hls://{0}' instead".format(self.stream.url))
+            raise StreamError(f"Attempted to play a variant playlist, use 'hls://{self.stream.url}' instead")
 
         if playlist.iframes_only:
-            raise StreamError("Streams containing I-frames only is not playable")
+            raise StreamError("Streams containing I-frames only are not playable")
 
         media_sequence = playlist.media_sequence or 0
         sequences = [Sequence(media_sequence + i, s)
@@ -322,6 +321,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
             self.reload_playlist()
         except StreamError as err:
             log.error(f'{err}')
+            self.reader.close()
             return
 
         if self.playlist_end is None:


Test patch:
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
index 364939d4..ca0d119a 100644
--- a/tests/mixins/stream_hls.py
+++ b/tests/mixins/stream_hls.py
@@ -13,6 +13,7 @@ from streamlink.stream.hls import HLSStream, HLSStreamWriter as _HLSStreamWriter
 TIMEOUT_AWAIT_READ = 5
 TIMEOUT_AWAIT_READ_ONCE = 5
 TIMEOUT_AWAIT_WRITE = 5
+TIMEOUT_AWAIT_CLOSE = 5
 
 
 class HLSItemBase:
@@ -205,14 +206,8 @@ class TestMixinStreamHLS(unittest.TestCase):
         super().tearDown()
 
         # close read thread and make sure that all threads have terminated before moving on
-        thread = self.thread
-        thread.reader.close()
-        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
-            thread.reader.writer.write_wait.set()
-        thread.read_wait.set()
-        thread.reader.writer.join()
-        thread.reader.worker.join()
-        thread.join()
+        self.close()
+        self.await_close()
 
         self.mocker.stop()
         self.mocks.clear()
@@ -236,6 +231,12 @@ class TestMixinStreamHLS(unittest.TestCase):
             segments = segments.values()
         return b"".join([getattr(segment, prop) for segment in segments if cond is None or cond(segment)])
 
+    def await_close(self, timeout=TIMEOUT_AWAIT_CLOSE):
+        thread = self.thread
+        thread.reader.writer.join(timeout)
+        thread.reader.worker.join(timeout)
+        thread.join(timeout)
+
     # make one write call on the write thread and wait until it has finished
     def await_write(self, write_calls=1, timeout=TIMEOUT_AWAIT_WRITE):
         writer = self.thread.reader.writer
@@ -296,3 +297,10 @@ class TestMixinStreamHLS(unittest.TestCase):
     def start(self):
         self.thread.reader.open()
         self.thread.start()
+
+    def close(self):
+        thread = self.thread
+        thread.reader.close()
+        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
+            thread.reader.writer.write_wait.set()
+        thread.read_wait.set()
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index 7d7fb50b..f7b5b4a5 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -1,4 +1,5 @@
 import os
+import typing
 import unittest
 from threading import Event
 from unittest.mock import Mock, call, patch
@@ -359,6 +360,65 @@ class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
         self.assertEqual(time, 6, "sets reload time to 6 seconds when no data is available")
 
 
+@patch("streamlink.stream.hls.log")
+@patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
+class TestHlsPlaylistParseErrors(TestMixinStreamHLS, unittest.TestCase):
+    __stream__ = EventedHLSStream
+
+    class FakePlaylist(typing.NamedTuple):
+        is_master: bool = False
+        iframes_only: bool = False
+
+    class InvalidPlaylist(Playlist):
+        def build(self, *args, **kwargs):
+            return "invalid"
+
+    def test_generic(self, mock_log):
+        self.subject([self.InvalidPlaylist()])
+        self.assertEqual(self.await_read(read_all=True), b"")
+        self.await_close()
+        self.assertTrue(self.thread.reader.buffer.closed, "Closes the stream on initial playlist parsing error")
+        self.assertEqual(mock_log.debug.mock_calls, [call("Reloading playlist")])
+        self.assertEqual(mock_log.error.mock_calls, [call("Missing #EXTM3U header")])
+
+    def test_reload(self, mock_log):
+        thread, segments = self.subject([
+            Playlist(1, [Segment(0)]),
+            self.InvalidPlaylist(),
+            self.InvalidPlaylist(),
+            Playlist(2, [Segment(2)], end=True)
+        ])
+        self.await_write(2)
+        data = self.await_read(read_all=True)
+        self.assertEqual(data, self.content(segments))
+        self.close()
+        self.await_close()
+        self.assertEqual(mock_log.warning.mock_calls, [
+            call("Failed to reload playlist: Missing #EXTM3U header"),
+            call("Failed to reload playlist: Missing #EXTM3U header")
+        ])
+
+    @patch("streamlink.stream.hls.HLSStreamWorker._reload_playlist", Mock(return_value=FakePlaylist(is_master=True)))
+    def test_is_master(self, mock_log):
+        self.subject([Playlist()])
+        self.assertEqual(self.await_read(read_all=True), b"")
+        self.await_close()
+        self.assertTrue(self.thread.reader.buffer.closed, "Closes the stream on initial playlist parsing error")
+        self.assertEqual(mock_log.debug.mock_calls, [call("Reloading playlist")])
+        self.assertEqual(mock_log.error.mock_calls, [
+            call(f"Attempted to play a variant playlist, use 'hls://{self.stream.url}' instead")
+        ])
+
+    @patch("streamlink.stream.hls.HLSStreamWorker._reload_playlist", Mock(return_value=FakePlaylist(iframes_only=True)))
+    def test_iframes_only(self, mock_log):
+        self.subject([Playlist()])
+        self.assertEqual(self.await_read(read_all=True), b"")
+        self.await_close()
+        self.assertTrue(self.thread.reader.buffer.closed, "Closes the stream on initial playlist parsing error")
+        self.assertEqual(mock_log.debug.mock_calls, [call("Reloading playlist")])
+        self.assertEqual(mock_log.error.mock_calls, [call("Streams containing I-frames only are not playable")])
+
+
 @patch('streamlink.stream.hls.FFMPEGMuxer.is_usable', Mock(return_value=True))
 class TestHlsExtAudio(unittest.TestCase):
     @property