Repository: streamlink/streamlink
Pull request #3209 (2020-09-28 07:05:23)

tests: clean up existing HLS stream tests
Sorry for another wall of text, but this is a fix / cleanup for all existing HLS integration tests which I've mentioned in #3194. Please bear with me, because that are quite a lot of changes, although not too difficult and also split up into multiple commits. Except for a few cases, the logic of the individual tests was not changed and just cleaned up / simplified.

Description of the individual commits:

1. Adds the `TestMixinStreamHLS` class and related ones for simplification of all HLS integration tests. The module was added to the `tests.mixins.stream_hls` import path, but I'm not sure if it's a good choice or if it breaks any conventions. If so, this needs to be moved elsewhere.

   The mixin provides methods for creating a fake HLS stream, mocking the requests, creating a dedicated read thread which can be controlled from the main thread and it adds proper tear-down methods, so that all threads properly terminate first before continuing with the next test. Its logic is partly taken from the previous implementation of the `hls_filtered` tests.

   Instead of manually building the fake HLS streams, they can now be created via extensible `Playlist`, `Tag` and `Segment` classes. This is rather basic right now and doesn't cover every existing HLS tag, but it provides a much simpler interface which makes the test more clean and easier to write.

2. Refactors the `streams.test_hls_filtered` test and makes use of the new mixin. The logic for stepping through each write call individually (due to the test's hls-timeout value of 0) is not part of the mixin and has been kept here, as it's not needed anywhere else.

   The changes also resolve the sloppy `time.sleep(0.1)` workaround from the previous PR by mocking the `filter_event.wait()` call of the reader thread with another `threading.Event.wait()` call on the main thread. The solution is a bit weird unfortunately, because Python doesn't support wait-until-executed Mocks (there's a stale PR on cpython for that exact thing).

3. Refactors the `plugins.test_twitch` tests. Not much to see here apart from the removed unnecessary master playlist setup, the added test assertion messages and overall cleanup.

4. Refactors the `TestHlsPlaylistReloadTime` tests in `streams.test_hls` and makes use of the new mixin. This should fix the test coverage wonkiness, because the HLS streams previously kept running for half a second or so, still trying to make network requests.

5. Separates the `HLSStream.parse_variant_playlist` tests from the existing HLS tests in `streams.test_hls`. Since the master playlist has nothing to do with those stream tests, it needs to be tested in its own testcase class.

   I've added a few more assertions here which required me to change the return value of the `HLSStream.parse_variant_playlist` method to be an `OrderedDict` instead of a `dict`, because there was a difference between py2 and py3 when not sorting the streams (like the plugin API does). Also the usual unicode nonsense in py2.

6. Separates the regular and encrypted HLS stream tests in `streams.test_hls` and makes use of the new mixin.

   I've added a couple more test assertions and changed the parameters a bit.

   The encryption key override test has exposed a minor issue with the `{query}` variable and the quotation mark had to be added to the URL override template.


Patch:
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index 2a69ca96..1ade39a9 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -2,7 +2,7 @@ import logging
 import re
 import struct
 
-from collections import defaultdict, namedtuple
+from collections import defaultdict, namedtuple, OrderedDict
 from Crypto.Cipher import AES
 from requests.exceptions import ChunkedEncodingError
 
@@ -429,7 +429,7 @@ class HLSStream(HTTPStream):
         except ValueError as err:
             raise IOError("Failed to parse playlist: {0}".format(err))
 
-        streams = {}
+        streams = OrderedDict()
         for playlist in filter(lambda p: not p.is_iframe, parser.playlists):
             names = dict(name=None, pixels=None, bitrate=None)
             audio_streams = []


Test patch:
diff --git a/tests/mixins/__init__.py b/tests/mixins/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
new file mode 100644
index 00000000..ec8ace24
--- /dev/null
+++ b/tests/mixins/stream_hls.py
@@ -0,0 +1,215 @@
+from binascii import hexlify
+from collections import OrderedDict
+from functools import partial
+from threading import Event, Thread
+
+import requests_mock
+import unittest
+
+from streamlink import Streamlink
+from streamlink.stream.hls import HLSStream
+
+
+class HLSItemBase(object):
+    path = ""
+
+    def url(self, namespace):
+        return "http://mocked/{namespace}/{path}".format(namespace=namespace, path=self.path)
+
+
+class Playlist(HLSItemBase):
+    path = "playlist.m3u8"
+
+    def __init__(self, mediasequence=None, segments=None, end=False, targetduration=0, version=7):
+        self.items = [
+            Tag("EXTM3U"),
+            Tag("EXT-X-VERSION", int(version)),
+            Tag("EXT-X-TARGETDURATION", int(targetduration))
+        ]
+        if mediasequence is not None:  # pragma: no branch
+            self.items.append(Tag("EXT-X-MEDIA-SEQUENCE", int(mediasequence)))
+        self.items += segments or []
+        if end:
+            self.items.append(Tag("EXT-X-ENDLIST"))
+
+    def build(self, *args, **kwargs):
+        return "\n".join([item.build(*args, **kwargs) for item in self.items])
+
+
+class Tag(HLSItemBase):
+    def __init__(self, name, attrs=None):
+        self.name = name
+        self.attrs = attrs
+
+    @classmethod
+    def val_quoted_string(cls, value):
+        return "\"{0}\"".format(value)
+
+    @classmethod
+    def val_hex(cls, value):
+        return "0x{0}".format(hexlify(value).decode("ascii"))
+
+    def build(self, *args, **kwargs):
+        attrs = None
+        if type(self.attrs) == dict:
+            attrs = ",".join([
+                "{0}={1}".format(key, value(self, *args, **kwargs) if callable(value) else value)
+                for (key, value) in self.attrs.items()
+            ])
+        elif self.attrs is not None:
+            attrs = str(self.attrs)
+
+        return "#{name}{attrs}".format(name=self.name, attrs=":{0}".format(attrs) if attrs else "")
+
+
+class Segment(HLSItemBase):
+    def __init__(self, num, title=None, duration=None, path_relative=True):
+        self.num = int(num or 0)
+        self.title = str(title or "")
+        self.duration = float(duration or 1)
+        self.path_relative = bool(path_relative)
+        self.content = "[{0}]".format(self.num).encode("ascii")
+
+    @property
+    def path(self):
+        return "segment{0}.ts".format(self.num)
+
+    def build(self, namespace):
+        return "#EXTINF:{duration:.3f},{title}\n{path}".format(
+            duration=self.duration,
+            title=self.title,
+            path=self.path if self.path_relative else self.url(namespace)
+        )
+
+
+class HLSStreamReadThread(Thread):
+    """
+    Run the reader on a separate thread, so that each read can be controlled from within the main thread
+    """
+    def __init__(self, session, stream, *args, **kwargs):
+        """
+        :param Streamlink session:
+        :param HLSStream stream:
+        """
+        Thread.__init__(self, *args, **kwargs)
+        self.daemon = True
+
+        self.read_wait = Event()
+        self.read_done = Event()
+        self.read_all = False
+        self.data = []
+        self.error = None
+
+        self.session = session
+        self.stream = stream
+        self.reader = stream.open()
+
+    def run(self):
+        while not self.reader.buffer.closed:
+            # only read once per step
+            self.read_wait.wait()
+            self.read_wait.clear()
+
+            try:
+                # don't read again during teardown
+                # if there is data left, close() was called manually, and it needs to be read
+                if self.reader.buffer.closed and self.reader.buffer.length == 0:
+                    return
+
+                if self.read_all:
+                    self.data += list(iter(partial(self.reader.read, -1), b""))
+                    return
+
+                self.data.append(self.reader.read(-1))
+            except IOError as err:
+                self.error = err
+                return
+            finally:
+                # notify main thread that reading has finished
+                self.read_done.set()
+
+    def reset(self):
+        self.data[:] = []
+        self.error = None
+
+
+class TestMixinStreamHLS(unittest.TestCase):
+    __stream__ = HLSStream
+    __readthread__ = HLSStreamReadThread
+
+    def __init__(self, *args, **kwargs):
+        super(TestMixinStreamHLS, self).__init__(*args, **kwargs)
+        self.mocker = requests_mock.Mocker()
+        self.mocks = {}
+        self.session = None
+        self.stream = None
+        self.thread = None
+
+    def setUp(self):
+        super(TestMixinStreamHLS, self).setUp()
+        self.mocker.start()
+
+    def tearDown(self):
+        super(TestMixinStreamHLS, self).tearDown()
+        self.close_thread()
+        self.mocker.stop()
+        self.mocks.clear()
+        self.session = None
+        self.stream = None
+        self.thread = None
+
+    def mock(self, method, url, *args, **kwargs):
+        self.mocks[url] = self.mocker.request(method, url, *args, **kwargs)
+
+    def called(self, item):
+        return self.mocks[self.url(item)].called
+
+    def url(self, item):
+        return item.url(self.id())
+
+    def content(self, segments, prop="content", cond=None):
+        return b"".join([getattr(segment, prop) for segment in segments.values() if cond is None or cond(segment)])
+
+    # close read thread and make sure that all threads have terminated before moving on
+    def close_thread(self):
+        thread = self.thread
+        thread.reader.close()
+        thread.read_wait.set()
+        thread.reader.writer.join()
+        thread.reader.worker.join()
+        thread.join()
+
+    # make one read call on the read thread and wait until it has finished
+    def await_read(self, read_all=False, timeout=5):
+        thread = self.thread
+        thread.read_all = read_all
+        thread.read_wait.set()
+        thread.read_done.wait(timeout)
+        thread.read_done.clear()
+
+        try:
+            if thread.error:
+                raise thread.error
+            return b"".join(thread.data)
+        finally:
+            thread.reset()
+
+    def get_session(self, options=None, *args, **kwargs):
+        return Streamlink(options)
+
+    # set up HLS responses, create the session and read thread and start it
+    def subject(self, playlists, options=None, streamoptions=None, threadoptions=None, *args, **kwargs):
+        # filter out tags and duplicate segments between playlist responses while keeping index order
+        segments_all = [item for playlist in playlists for item in playlist.items if isinstance(item, Segment)]
+        segments = OrderedDict([(segment.num, segment) for segment in segments_all])
+
+        self.mock("GET", self.url(playlists[0]), [{"text": pl.build(self.id())} for pl in playlists])
+        for segment in segments.values():
+            self.mock("GET", self.url(segment), content=segment.content)
+
+        self.session = self.get_session(options, *args, **kwargs)
+        self.stream = self.__stream__(self.session, self.url(playlists[0]), **(streamoptions or {}))
+        self.thread = self.__readthread__(self.session, self.stream, **(threadoptions or {}))
+        self.thread.start()
+
+        return self.thread, segments
diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
index d84ada7a..3056c4cc 100644
--- a/tests/plugins/test_twitch.py
+++ b/tests/plugins/test_twitch.py
@@ -1,13 +1,22 @@
+import requests_mock
 import unittest
-from functools import partial
 
+from tests.mixins.stream_hls import Playlist, Segment as Segment, TestMixinStreamHLS
+from tests.mock import MagicMock, call, patch
+
+from streamlink import Streamlink
 from streamlink.plugins.twitch import Twitch, TwitchHLSStream
 
-import requests_mock
-from tests.mock import MagicMock, call, patch
 
-from streamlink.session import Streamlink
-from tests.resources import text
+class SegmentAd(Segment):
+    def __init__(self, *args, **kwargs):
+        super(SegmentAd, self).__init__(*args, **kwargs)
+        self.title = "Amazon|123456789"
+
+
+class SegmentPrefetch(Segment):
+    def build(self, namespace):
+        return "#EXT-X-TWITCH-PREFETCH:{0}".format(self.url(namespace))
 
 
 class TestPluginTwitch(unittest.TestCase):
@@ -31,218 +40,166 @@ class TestPluginTwitch(unittest.TestCase):
 
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", MagicMock(return_value=True))
-class TestTwitchHLSStream(unittest.TestCase):
-    url_master = "http://mocked/path/master.m3u8"
-    url_playlist = "http://mocked/path/playlist.m3u8"
-    url_segment = "http://mocked/path/stream{0}.ts"
-
-    segment = "#EXTINF:1.000,\nstream{0}.ts\n"
-    segment_ad = "#EXTINF:1.000,Amazon|123456789\nstream{0}.ts\n"
-    prefetch = "#EXT-X-TWITCH-PREFETCH:{0}\n"
-
-    def getMasterPlaylist(self):
-        with text("hls/test_master.m3u8") as pl:
-            return pl.read()
-
-    def getPlaylist(self, media_sequence, items, ads=False, prefetch=None, end=False):
-        playlist = """
-#EXTM3U
-#EXT-X-VERSION:5
-#EXT-X-TARGETDURATION:1
-#EXT-X-MEDIA-SEQUENCE:{0}
-""".format(media_sequence)
-
-        segment = self.segment if not ads else self.segment_ad
-        for item in items:
-            playlist += segment.format(item)
-        for item in prefetch or []:
-            playlist += self.prefetch.format(self.url_segment.format(item))
-
-        if end:
-            playlist += "#EXT-X-ENDLIST\n"
-
-        return playlist
-
-    def getSegments(self, num):
-        return ["[{0}]".format(i).encode("ascii") for i in range(num)]
-
-    def start_streamlink(self, disable_ads=False, low_latency=False, kwargs=None):
-        kwargs = kwargs or {}
-        streamlink = Streamlink()
-
-        streamlink.set_option("hls-live-edge", 4)
-        streamlink.set_plugin_option("twitch", "disable-ads", disable_ads)
-        streamlink.set_plugin_option("twitch", "low-latency", low_latency)
-
-        masterStream = TwitchHLSStream.parse_variant_playlist(streamlink, self.url_master, **kwargs)
-        stream = masterStream["1080p (source)"].open()
-        data = b"".join(iter(partial(stream.read, 8192), b""))
-        stream.close()
-        return streamlink, data
-
-    def mock(self, mocked, method, url, *args, **kwargs):
-        mocked[url] = method(url, *args, **kwargs)
-
-    def get_result(self, segments, playlists, **kwargs):
-        mocked = {}
-        with requests_mock.Mocker() as mock:
-            self.mock(mocked, mock.get, self.url_master, text=self.getMasterPlaylist())
-            self.mock(mocked, mock.get, self.url_playlist, [{"text": p} for p in playlists])
-            for i, segment in enumerate(segments):
-                self.mock(mocked, mock.get, self.url_segment.format(i), content=segment)
-            streamlink, data = self.start_streamlink(**kwargs)
-            return streamlink, data, mocked
+class TestTwitchHLSStream(TestMixinStreamHLS, unittest.TestCase):
+    __stream__ = TwitchHLSStream
 
-    @patch("streamlink.plugins.twitch.log")
-    def test_hls_disable_ads_has_preroll(self, mock_logging):
-        segments = self.getSegments(6)
-        playlists = [
-            self.getPlaylist(0, [0, 1], ads=True),
-            self.getPlaylist(2, [2, 3], ads=True),
-            self.getPlaylist(4, [4, 5], ads=False, end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=True, low_latency=False)
+    def get_session(self, options=None, disable_ads=False, low_latency=False):
+        session = super(TestTwitchHLSStream, self).get_session(options)
+        session.set_option("hls-live-edge", 4)
+        session.set_plugin_option("twitch", "disable-ads", disable_ads)
+        session.set_plugin_option("twitch", "low-latency", low_latency)
+
+        return session
 
-        self.assertEqual(result, b"".join(segments[4:6]))
-        for i in range(0, 6):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_disable_ads_has_preroll(self, mock_log):
+        thread, segments = self.subject([
+            Playlist(0, [SegmentAd(0), SegmentAd(1)]),
+            Playlist(2, [SegmentAd(2), SegmentAd(3)]),
+            Playlist(4, [Segment(4), Segment(5)], end=True)
+        ], disable_ads=True, low_latency=False)
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments, cond=lambda s: s.num >= 4),
+            "Filters out preroll ad segments"
+        )
+        self.assertTrue(all([self.called(s) for s in segments.values()]), "Downloads all segments")
+        self.assertEqual(mock_log.info.mock_calls, [
             call("Will skip ad segments"),
             call("Waiting for pre-roll ads to finish, be patient")
         ])
-        self.assertEqual(mock_logging.info.call_count, 2)
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_disable_ads_has_midstream(self, mock_logging):
-        segments = self.getSegments(6)
-        playlists = [
-            self.getPlaylist(0, [0, 1], ads=False),
-            self.getPlaylist(2, [2, 3], ads=True),
-            self.getPlaylist(4, [4, 5], ads=False, end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=True, low_latency=False)
-
-        self.assertEqual(result, b"".join(segments[0:2]) + b"".join(segments[4:6]))
-        for i in range(0, 6):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([
+    def test_hls_disable_ads_has_midstream(self, mock_log):
+        thread, segments = self.subject([
+            Playlist(0, [Segment(0), Segment(1)]),
+            Playlist(2, [SegmentAd(2), SegmentAd(3)]),
+            Playlist(4, [Segment(4), Segment(5)], end=True)
+        ], disable_ads=True, low_latency=False)
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments, cond=lambda s: s.num != 2 and s.num != 3),
+            "Filters out mid-stream ad segments"
+        )
+        self.assertTrue(all([self.called(s) for s in segments.values()]), "Downloads all segments")
+        self.assertEqual(mock_log.info.mock_calls, [
             call("Will skip ad segments")
         ])
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_no_disable_ads_has_preroll(self, mock_logging):
-        segments = self.getSegments(4)
-        playlists = [
-            self.getPlaylist(0, [0, 1], ads=True),
-            self.getPlaylist(2, [2, 3], ads=False, end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=False, low_latency=False)
-
-        self.assertEqual(result, b"".join(segments[0:4]))
-        for i in range(0, 4):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([])
+    def test_hls_no_disable_ads_has_preroll(self, mock_log):
+        thread, segments = self.subject([
+            Playlist(0, [SegmentAd(0), SegmentAd(1)]),
+            Playlist(2, [Segment(2), Segment(3)], end=True)
+        ], disable_ads=False, low_latency=False)
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments),
+            "Doesn't filter out segments"
+        )
+        self.assertTrue(all([self.called(s) for s in segments.values()]), "Downloads all segments")
+        self.assertEqual(mock_log.info.mock_calls, [], "Doesn't log anything")
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_low_latency_has_prefetch(self, mock_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[4, 5]),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[8, 9], end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=False, low_latency=True)
-
-        self.assertEqual(2, streamlink.options.get("hls-live-edge"))
-        self.assertEqual(True, streamlink.options.get("hls-segment-stream-data"))
-
-        self.assertEqual(result, b"".join(segments[4:10]))
-        for i in range(0, 3):
-            self.assertFalse(mocked[self.url_segment.format(i)].called, i)
-        for i in range(4, 9):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([
+    def test_hls_low_latency_has_prefetch(self, mock_log):
+        thread, 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)
+        ], disable_ads=False, low_latency=True)
+
+        self.assertEqual(2, self.session.options.get("hls-live-edge"))
+        self.assertEqual(True, self.session.options.get("hls-segment-stream-data"))
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments, cond=lambda s: s.num >= 4),
+            "Skips first four segments due to reduced live-edge"
+        )
+        self.assertFalse(any([self.called(s) for s in segments.values() if s.num < 4]), "Doesn't download old segments")
+        self.assertTrue(all([self.called(s) for s in segments.values() if s.num >= 4]), "Downloads all remaining segments")
+        self.assertEqual(mock_log.info.mock_calls, [
             call("Low latency streaming (HLS live edge: 2)")
         ])
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_no_low_latency_has_prefetch(self, mock_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[4, 5]),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[8, 9], end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=False, low_latency=False)
-
-        self.assertEqual(4, streamlink.options.get("hls-live-edge"))
-        self.assertEqual(False, streamlink.options.get("hls-segment-stream-data"))
-
-        self.assertEqual(result, b"".join(segments[0:8]))
-        for i in range(0, 7):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        for i in range(8, 9):
-            self.assertFalse(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([])
+    def test_hls_no_low_latency_has_prefetch(self, mock_log):
+        thread, 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)
+        ], disable_ads=False, low_latency=False)
+
+        self.assertEqual(4, self.session.options.get("hls-live-edge"))
+        self.assertEqual(False, self.session.options.get("hls-segment-stream-data"))
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments, cond=lambda s: s.num < 8),
+            "Ignores prefetch segments"
+        )
+        self.assertTrue(all([self.called(s) for s in segments.values() if s.num <= 7]), "Ignores prefetch segments")
+        self.assertFalse(any([self.called(s) for s in segments.values() if s.num > 7]), "Ignores prefetch segments")
+        self.assertEqual(mock_log.info.mock_calls, [], "Doesn't log anything")
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_low_latency_no_prefetch(self, mock_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[]),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[], end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=False, low_latency=True)
+    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)
+        ], disable_ads=False, low_latency=True)
 
-        self.assertTrue(streamlink.get_plugin_option("twitch", "low-latency"))
-        self.assertFalse(streamlink.get_plugin_option("twitch", "disable-ads"))
+        self.assertTrue(self.session.get_plugin_option("twitch", "low-latency"))
+        self.assertFalse(self.session.get_plugin_option("twitch", "disable-ads"))
 
-        mock_logging.info.assert_has_calls([
+        self.await_read(read_all=True)
+        self.assertEqual(mock_log.info.mock_calls, [
             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_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[], ads=True),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[8, 9], ads=False, end=True)
-        ]
-        streamlink, result, mocked = self.get_result(segments, playlists, disable_ads=False, low_latency=True)
-
-        self.assertEqual(result, b"".join(segments[2:10]))
-        for i in range(0, 2):
-            self.assertFalse(mocked[self.url_segment.format(i)].called, i)
-        for i in range(2, 10):
-            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
-        mock_logging.info.assert_has_calls([
+    def test_hls_low_latency_has_prefetch_has_preroll(self, mock_log):
+        thread, segments = self.subject([
+            Playlist(0, [SegmentAd(0), SegmentAd(1), SegmentAd(2), SegmentAd(3)]),
+            Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True)
+        ], disable_ads=False, low_latency=True)
+
+        self.assertEqual(
+            self.await_read(read_all=True),
+            self.content(segments, cond=lambda s: s.num > 1),
+            "Skips first two segments due to reduced live-edge"
+        )
+        self.assertFalse(any([self.called(s) for s in segments.values() if s.num < 2]), "Skips first two preroll segments")
+        self.assertTrue(all([self.called(s) for s in segments.values() if s.num >= 2]), "Downloads all remaining segments")
+        self.assertEqual(mock_log.info.mock_calls, [
             call("Low latency streaming (HLS live edge: 2)")
         ])
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_low_latency_has_prefetch_disable_ads_has_preroll(self, mock_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[], ads=True),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[8, 9], ads=False, end=True)
-        ]
-        self.get_result(segments, playlists, disable_ads=True, low_latency=True)
-
-        mock_logging.info.assert_has_calls([
+    def test_hls_low_latency_has_prefetch_disable_ads_has_preroll(self, mock_log):
+        self.subject([
+            Playlist(0, [SegmentAd(0), SegmentAd(1), SegmentAd(2), SegmentAd(3)]),
+            Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7), SegmentPrefetch(8), SegmentPrefetch(9)], end=True)
+        ], disable_ads=True, low_latency=True)
+
+        self.await_read(read_all=True)
+        self.assertEqual(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")
         ])
 
     @patch("streamlink.plugins.twitch.log")
-    def test_hls_low_latency_no_prefetch_disable_ads_has_preroll(self, mock_logging):
-        segments = self.getSegments(10)
-        playlists = [
-            self.getPlaylist(0, [0, 1, 2, 3], prefetch=[], ads=True),
-            self.getPlaylist(4, [4, 5, 6, 7], prefetch=[], ads=False, end=True)
-        ]
-        self.get_result(segments, playlists, disable_ads=True, low_latency=True)
-
-        mock_logging.info.assert_has_calls([
+    def test_hls_low_latency_no_prefetch_disable_ads_has_preroll(self, mock_log):
+        self.subject([
+            Playlist(0, [SegmentAd(0), SegmentAd(1), SegmentAd(2), SegmentAd(3)]),
+            Playlist(4, [Segment(4), Segment(5), Segment(6), Segment(7)], end=True)
+        ], disable_ads=True, low_latency=True)
+
+        self.await_read(read_all=True)
+        self.assertEqual(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"),
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index b131df56..37381d5c 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -1,20 +1,17 @@
-import itertools
-import logging
 import os
 import unittest
-from binascii import hexlify
-from functools import partial
 
+from Crypto.Cipher import AES
 import pytest
 import requests_mock
-from Crypto.Cipher import AES
-from mock import patch, Mock
 
-from streamlink.session import Streamlink
-from streamlink.stream import hls
+from tests.mixins.stream_hls import Playlist, Tag, Segment, TestMixinStreamHLS
+from tests.mock import patch, Mock
 from tests.resources import text
 
-log = logging.getLogger(__name__)
+from streamlink.compat import str
+from streamlink.session import Streamlink
+from streamlink.stream import hls
 
 
 def pkcs7_encode(data, keySize):
@@ -28,203 +25,175 @@ def encrypt(data, key, iv):
     return encrypted_data
 
 
-@patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
-class TestHLS(unittest.TestCase):
-    """
-    Test that when invoked for the command line arguments are parsed as expected
-    """
-    mediaSequence = 1651
-
-    def getMasterPlaylist(self):
-        with text("hls/test_master.m3u8") as pl:
-            return pl.read()
+class TagKey(Tag):
+    path = "encryption.key"
 
-    def getPlaylist(self, aesIv, streamNameTemplate):
-        playlist = """
-#EXTM3U
-#EXT-X-VERSION:5
-#EXT-X-TARGETDURATION:1
-#ID3-EQUIV-TDTG:2018-01-01T18:20:05
-#EXT-X-MEDIA-SEQUENCE:{0}
-#EXT-X-TWITCH-ELAPSED-SECS:3367.800
-#EXT-X-TWITCH-TOTAL-SECS:3379.943
-""".format(self.mediaSequence)
-
-        playlistEnd = ""
-        if aesIv is not None:
-            ext_x_key = "#EXT-X-KEY:METHOD=AES-128,URI=\"{uri}\",IV=0x{iv},KEYFORMAT=identity,KEYFORMATVERSIONS=1\n"
-            playlistEnd = playlistEnd + ext_x_key.format(uri="encryption_key.key", iv=hexlify(aesIv).decode("UTF-8"))
-
-        for i in range(4):
-            playlistEnd = playlistEnd + "#EXTINF:1.000,\n{0}\n".format(streamNameTemplate.format(i))
-            self.mediaSequence += 1
-
-        return playlist + playlistEnd
-
-    def start_streamlink(self, masterPlaylist, hls_segment_key_uri=None, kwargs=None):
-        kwargs = kwargs or {}
-        log.info("Executing streamlink")
-        streamlink = Streamlink()
+    def __init__(self, method="NONE", uri=None, iv=None, keyformat=None, keyformatversions=None):
+        attrs = {"METHOD": method}
+        if uri is not False:  # pragma: no branch
+            attrs.update({"URI": lambda tag, namespace: tag.val_quoted_string(tag.url(namespace))})
+        if iv is not None:  # pragma: no branch
+            attrs.update({"IV": self.val_hex(iv)})
+        if keyformat is not None:  # pragma: no branch
+            attrs.update({"KEYFORMAT": self.val_quoted_string(keyformat)})
+        if keyformatversions is not None:  # pragma: no branch
+            attrs.update({"KEYFORMATVERSIONS": self.val_quoted_string(keyformatversions)})
+        super(TagKey, self).__init__("EXT-X-KEY", attrs)
+        self.uri = uri
+
+    def url(self, namespace):
+        return self.uri.format(namespace=namespace) if self.uri else super(TagKey, self).url(namespace)
 
-        # Set to default value to avoid a test fail if the default change
-        streamlink.set_option("hls-live-edge", 3)
-        streamlink.set_option("hls-segment-key-uri", hls_segment_key_uri)
-
-        masterStream = hls.HLSStream.parse_variant_playlist(streamlink, masterPlaylist, **kwargs)
-        stream = masterStream["1080p (source)"].open()
-        data = b"".join(iter(partial(stream.read, 8192), b""))
-        stream.close()
-        log.info("End of streamlink execution")
-        return data
-
-    def test_hls_non_encrypted(self):
-        streams = [os.urandom(1024) for _ in range(4)]
-        masterPlaylist = self.getMasterPlaylist()
-        playlist = self.getPlaylist(None, "stream{0}.ts") + "#EXT-X-ENDLIST\n"
+
+class SegmentEnc(Segment):
+    def __init__(self, num, key, iv, *args, **kwargs):
+        super(SegmentEnc, self).__init__(num, *args, **kwargs)
+        self.content_plain = self.content
+        self.content = encrypt(self.content, key, iv)
+
+
+class TestHLSVariantPlaylist(unittest.TestCase):
+    @classmethod
+    def get_master_playlist(cls, playlist):
+        with text(playlist) as pl:
+            return pl.read()
+
+    def subject(self, playlist, options=None):
         with requests_mock.Mocker() as mock:
-            mock.get("http://mocked/path/master.m3u8", text=masterPlaylist)
-            mock.get("http://mocked/path/playlist.m3u8", text=playlist)
-            for i, stream in enumerate(streams):
-                mock.get("http://mocked/path/stream{0}.ts".format(i), content=stream)
+            url = "http://mocked/{0}/master.m3u8".format(self.id())
+            content = self.get_master_playlist(playlist)
+            mock.get(url, text=content)
 
-            # Start streamlink on the generated stream
-            streamlinkResult = self.start_streamlink("http://mocked/path/master.m3u8",
-                                                     kwargs={'start_offset': 1, 'duration': 1})
+            session = Streamlink(options)
 
-        # Check result, each segment is 1 second, with duration=1 only one segment should be returned
-        expectedResult = b''.join(streams[1:2])
-        self.assertEqual(streamlinkResult, expectedResult)
+            return hls.HLSStream.parse_variant_playlist(session, url)
+
+    def test_variant_playlist(self):
+        streams = self.subject("hls/test_master.m3u8")
+        self.assertEqual(
+            [str(key) for key in streams.keys()],
+            [u"720p", u"720p_alt", u"480p", u"360p", u"160p", u"1080p (source)", u"90k"],
+            "Finds all streams in master playlist"
+        )
+        self.assertTrue(
+            all([isinstance(stream, hls.HLSStream) for stream in streams.values()]),
+            "Returns HLSStream instances"
+        )
+
+
+@patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
+class TestHLSStream(TestMixinStreamHLS, unittest.TestCase):
+    def get_session(self, options=None, *args, **kwargs):
+        session = super(TestHLSStream, self).get_session(options)
+        session.set_option("hls-live-edge", 3)
+
+        return session
+
+    def test_offset_and_duration(self):
+        thread, segments = self.subject([
+            Playlist(1234, [Segment(0), Segment(1, duration=0.5), Segment(2, duration=0.5), Segment(3)], end=True)
+        ], streamoptions={"start_offset": 1, "duration": 1})
+
+        data = self.await_read(read_all=True)
+        self.assertEqual(data, self.content(segments, cond=lambda s: 0 < s.num < 3), "Respects the offset and duration")
+        self.assertTrue(all([self.called(s) for s in segments.values() if 0 < s.num < 3]), "Downloads second and third segment")
+        self.assertFalse(any([self.called(s) for s in segments.values() if 0 > s.num > 3]), "Skips other segments")
+
+
+@patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
+class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
+    def get_session(self, options=None, *args, **kwargs):
+        session = super(TestHLSStreamEncrypted, self).get_session(options)
+        session.set_option("hls-live-edge", 3)
+
+        return session
+
+    def gen_key(self, aes_key=None, aes_iv=None, method="AES-128", uri=None, keyformat="identity", keyformatversions=1):
+        aes_key = aes_key or os.urandom(16)
+        aes_iv = aes_iv or os.urandom(16)
+
+        key = TagKey(method=method, uri=uri, iv=aes_iv, keyformat=keyformat, keyformatversions=keyformatversions)
+        self.mock("GET", key.url(self.id()), content=aes_key)
+
+        return aes_key, aes_iv, key
 
     def test_hls_encrypted_aes128(self):
-        # Encryption parameters
-        aesKey = os.urandom(16)
-        aesIv = os.urandom(16)
-        # Generate stream data files
-        clearStreams = [os.urandom(1024) for i in range(4)]
-        encryptedStreams = [encrypt(clearStream, aesKey, aesIv) for clearStream in clearStreams]
-
-        masterPlaylist = self.getMasterPlaylist()
-        playlist1 = self.getPlaylist(aesIv, "stream{0}.ts.enc")
-        playlist2 = self.getPlaylist(aesIv, "stream2_{0}.ts.enc") + "#EXT-X-ENDLIST\n"
-
-        streamlinkResult = None
-        with requests_mock.Mocker() as mock:
-            mock.get("http://mocked/path/master.m3u8", text=masterPlaylist)
-            mock.get("http://mocked/path/playlist.m3u8", [{'text': playlist1}, {'text': playlist2}])
-            mock.get("http://mocked/path/encryption_key.key", content=aesKey)
-            for i, encryptedStream in enumerate(encryptedStreams):
-                mock.get("http://mocked/path/stream{0}.ts.enc".format(i), content=encryptedStream)
-            for i, encryptedStream in enumerate(encryptedStreams):
-                mock.get("http://mocked/path/stream2_{0}.ts.enc".format(i), content=encryptedStream)
+        aesKey, aesIv, key = self.gen_key()
 
-            # Start streamlink on the generated stream
-            streamlinkResult = self.start_streamlink("http://mocked/path/master.m3u8")
+        # noinspection PyTypeChecker
+        thread, segments = self.subject([
+            Playlist(0, [key] + [SegmentEnc(num, aesKey, aesIv) for num in range(0, 4)]),
+            Playlist(4, [key] + [SegmentEnc(num, aesKey, aesIv) for num in range(4, 8)], end=True)
+        ])
 
-        # Check result
-        # Live streams starts the last 3 segments from the playlist
-        expectedResult = b''.join(clearStreams[1:] + clearStreams)
-        self.assertEqual(streamlinkResult, expectedResult)
+        data = self.await_read(read_all=True)
+        expected = self.content(segments, prop="content_plain", cond=lambda s: s.num >= 1)
+        self.assertEqual(data, expected, "Decrypts the AES-128 identity stream")
+        self.assertTrue(self.called(key), "Downloads encryption key")
+        self.assertFalse(any([self.called(s) for s in segments.values() if s.num < 1]), "Skips first segment")
+        self.assertTrue(all([self.called(s) for s in segments.values() if s.num >= 1]), "Downloads all remaining segments")
 
     def test_hls_encrypted_aes128_key_uri_override(self):
-        aesKey = os.urandom(16)
-        aesIv = os.urandom(16)
+        aesKey, aesIv, key = self.gen_key(uri="http://real-mocked/{namespace}/encryption.key?foo=bar")
         aesKeyInvalid = bytes([ord(aesKey[i:i + 1]) ^ 0xFF for i in range(16)])
-        clearStreams = [os.urandom(1024) for i in range(4)]
-        encryptedStreams = [encrypt(clearStream, aesKey, aesIv) for clearStream in clearStreams]
+        _, __, key_invalid = self.gen_key(aesKeyInvalid, aesIv, uri="http://mocked/{namespace}/encryption.key?foo=bar")
 
-        masterPlaylist = self.getMasterPlaylist()
-        playlist1 = self.getPlaylist(aesIv, "stream{0}.ts.enc")
-        playlist2 = self.getPlaylist(aesIv, "stream2_{0}.ts.enc") + "#EXT-X-ENDLIST\n"
+        # noinspection PyTypeChecker
+        thread, segments = self.subject([
+            Playlist(0, [key_invalid] + [SegmentEnc(num, aesKey, aesIv) for num in range(0, 4)]),
+            Playlist(4, [key_invalid] + [SegmentEnc(num, aesKey, aesIv) for num in range(4, 8)], end=True)
+        ], options={"hls-segment-key-uri": "{scheme}://real-{netloc}{path}?{query}"})
 
-        mocked_key_uri_default = None
-        mocked_key_uri_override = None
-        streamlinkResult = None
-        with requests_mock.Mocker() as mock:
-            mock.get("http://mocked/path/master.m3u8", text=masterPlaylist)
-            mock.get("http://mocked/path/playlist.m3u8", [{'text': playlist1}, {'text': playlist2}])
-            mocked_key_uri_default = mock.get("http://mocked/path/encryption_key.key", content=aesKeyInvalid)
-            mocked_key_uri_override = mock.get("http://real-mocked/path/encryption_key.key", content=aesKey)
-            for i, encryptedStream in enumerate(encryptedStreams):
-                mock.get("http://mocked/path/stream{0}.ts.enc".format(i), content=encryptedStream)
-            for i, encryptedStream in enumerate(encryptedStreams):
-                mock.get("http://mocked/path/stream2_{0}.ts.enc".format(i), content=encryptedStream)
-
-            streamlinkResult = self.start_streamlink(
-                "http://mocked/path/master.m3u8",
-                hls_segment_key_uri="{scheme}://real-{netloc}{path}{query}"
-            )
-
-        self.assertFalse(mocked_key_uri_default.called)
-        self.assertTrue(mocked_key_uri_override.called)
-        expectedResult = b''.join(clearStreams[1:] + clearStreams)
-        self.assertEqual(streamlinkResult, expectedResult)
+        data = self.await_read(read_all=True)
+        expected = self.content(segments, prop="content_plain", cond=lambda s: s.num >= 1)
+        self.assertEqual(data, expected, "Decrypts stream from custom key")
+        self.assertFalse(self.called(key_invalid), "Skips encryption key")
+        self.assertTrue(self.called(key), "Downloads custom encryption key")
 
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
-class TestHlsPlaylistReloadTime(unittest.TestCase):
-    url_playlist = "http://mocked/path/playlist.m3u8"
-    url_segment = "http://mocked/path/stream{0}.ts"
-    segment = "#EXTINF:{duration},\nstream{num}.ts\n"
-
-    def getPlaylist(self, media_sequence, target_duration, offset, items):
-        return (
-            "#EXTM3U\n"
-            "#EXT-X-VERSION:5{target_duration}\n"
-            "#EXT-X-MEDIA-SEQUENCE:{media_sequence}\n"
-            "{items}\n"
-            "#EXT-X-ENDLIST\n"
-        ).format(
-            media_sequence=media_sequence,
-            target_duration="\n#EXT-X-TARGETDURATION:" + str(target_duration) if target_duration else "",
-            items="\n".join([self.segment.format(num=offset + i, duration=d) for i, d in enumerate(items)])
-        )
+@patch("streamlink.stream.hls.HLSStreamWriter.run", Mock(return_value=True))
+class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
+    segments = [Segment(0, "", 11), Segment(1, "", 7), Segment(2, "", 5), Segment(3, "", 3)]
 
-    def start_streamlink(self, data, reload_time):
-        streamlink = Streamlink()
-        streamlink.set_option("hls-playlist-reload-time", reload_time)
-        streamlink.set_option("hls-live-edge", 3)
-
-        streams = [b"" for i in itertools.chain.from_iterable([elem[3] for elem in data])]
-        playlists = [self.getPlaylist(*args) for args in data]
-
-        with requests_mock.Mocker() as mock:
-            mock.get(self.url_playlist, [{"text": p} for p in playlists])
-            for i, stream in enumerate(streams):
-                mock.get(self.url_segment.format(i), content=stream)
+    def get_session(self, options=None, reload_time=None, *args, **kwargs):
+        return super(TestHlsPlaylistReloadTime, self).get_session(dict(options or {}, **{
+            "hls-live-edge": 3,
+            "hls-playlist-reload-time": reload_time
+        }))
 
-            hlsstream = hls.HLSStream(streamlink, self.url_playlist)
-            reader = hlsstream.open()
-            reader.close()
+    def subject(self, *args, **kwargs):
+        thread, _ = super(TestHlsPlaylistReloadTime, self).subject(*args, **kwargs)
 
-            return reader
+        return thread.reader.worker.playlist_reload_time
 
     def test_hls_playlist_reload_time_default(self):
-        reader = self.start_streamlink([(0, 6, 1, [11, 7, 5, 3])], "default")
-        self.assertEqual(reader.worker.playlist_reload_time, 6)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="default")
+        self.assertEqual(time, 6, "default sets the reload time to the playlist's target duration")
 
     def test_hls_playlist_reload_time_segment(self):
-        reader = self.start_streamlink([(0, 6, 1, [11, 7, 5, 3])], "segment")
-        self.assertEqual(reader.worker.playlist_reload_time, 3)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="segment")
+        self.assertEqual(time, 3, "segment sets the reload time to the playlist's last segment")
 
     def test_hls_playlist_reload_time_live_edge(self):
-        reader = self.start_streamlink([(0, 6, 1, [11, 7, 5, 3])], "live-edge")
-        self.assertEqual(reader.worker.playlist_reload_time, 8)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="live-edge")
+        self.assertEqual(time, 8, "live-edge sets the reload time to the sum of the number of segments of the live-edge")
 
     def test_hls_playlist_reload_time_number(self):
-        reader = self.start_streamlink([(0, 6, 1, [11, 7, 5, 3])], "4")
-        self.assertEqual(reader.worker.playlist_reload_time, 4)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="4")
+        self.assertEqual(time, 4, "number values override the reload time")
 
     def test_hls_playlist_reload_time_number_invalid(self):
-        reader = self.start_streamlink([(0, 6, 1, [11, 7, 5, 3])], "0")
-        self.assertEqual(reader.worker.playlist_reload_time, 6)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="0")
+        self.assertEqual(time, 6, "invalid number values set the reload time to the playlist's targetduration")
 
     def test_hls_playlist_reload_time_no_target_duration(self):
-        reader = self.start_streamlink([(0, None, 1, [11, 7, 5, 3])], "default")
-        self.assertEqual(reader.worker.playlist_reload_time, 8)
+        time = self.subject([Playlist(0, self.segments, end=True, targetduration=0)], reload_time="default")
+        self.assertEqual(time, 8, "uses the live-edge sum if the playlist is missing the targetduration data")
 
     def test_hls_playlist_reload_time_no_data(self):
-        reader = self.start_streamlink([(0, None, 1, [])], "default")
-        self.assertEqual(reader.worker.playlist_reload_time, 15)
+        time = self.subject([Playlist(0, [], end=True, targetduration=0)], reload_time="default")
+        self.assertEqual(time, 15, "sets reload time to 15 seconds when no data is available")
 
 
 @patch('streamlink.stream.hls.FFMPEGMuxer.is_usable', Mock(return_value=True))
diff --git a/tests/streams/test_hls_filtered.py b/tests/streams/test_hls_filtered.py
index 72592757..545e8cf0 100644
--- a/tests/streams/test_hls_filtered.py
+++ b/tests/streams/test_hls_filtered.py
@@ -1,17 +1,23 @@
-import requests_mock
-from tests.mock import MagicMock, call, patch
+from threading import Event
+
 import unittest
 
-import itertools
-from textwrap import dedent
-from threading import Event, Thread
-from time import sleep
+from tests.mixins.stream_hls import Playlist, Segment, TestMixinStreamHLS
+from tests.mock import MagicMock, call, patch
 
-from streamlink.session import Streamlink
 from streamlink.stream.hls import HLSStream
 from streamlink.stream.hls_filtered import FilteredHLSStreamWriter, FilteredHLSStreamReader
 
 
+FILTERED = "filtered"
+
+
+class SegmentFiltered(Segment):
+    def __init__(self, *args, **kwargs):
+        super(SegmentFiltered, self).__init__(*args, **kwargs)
+        self.title = FILTERED
+
+
 class _TestSubjectFilteredHLSStreamWriter(FilteredHLSStreamWriter):
     def __init__(self, *args, **kwargs):
         super(_TestSubjectFilteredHLSStreamWriter, self).__init__(*args, **kwargs)
@@ -36,233 +42,168 @@ class _TestSubjectFilteredHLSReader(FilteredHLSStreamReader):
     __writer__ = _TestSubjectFilteredHLSStreamWriter
 
 
-class _TestSubjectReadThread(Thread):
-    """
-    Run the reader on a separate thread, so that each read can be controlled from within the main thread
-    """
-    def __init__(self):
-        Thread.__init__(self)
-        self.daemon = True
-
-        session = Streamlink()
-        session.set_option("hls-live-edge", 2)
-        session.set_option("hls-timeout", 0)
-        session.set_option("stream-timeout", 0)
-
-        self.read_wait = Event()
-        self.read_done = Event()
-        self.data = []
-        self.error = None
-
-        self.stream = HLSStream(session, TestFilteredHLSStream.url_playlist)
-        self.reader = _TestSubjectFilteredHLSReader(self.stream)
-        self.reader.open()
-
-    def run(self):
-        while not self.reader.buffer.closed:
-            # only read once per step
-            self.read_wait.wait()
-            self.read_wait.clear()
-
-            try:
-                # don't read again during teardown
-                # if there is data left, close() was called manually, and it needs to be read
-                if self.reader.buffer.closed and self.reader.buffer.length == 0:
-                    return
-                data = self.reader.read(-1)
-                self.data.append(data)
-            except IOError as err:
-                self.error = err
-                return
-            finally:
-                # notify main thread that reading has finished
-                self.read_done.set()
-
-    def await_write(self):
-        writer = self.reader.writer
-        # make one write call and wait until it has finished
-        writer.write_wait.set()
-        writer.write_done.wait()
-        writer.write_done.clear()
+class _TestSubjectFilteredHLSStream(HLSStream):
+    def open(self):
+        reader = _TestSubjectFilteredHLSReader(self)
+        reader.open()
 
-    def await_read(self):
-        # make one read call and wait until it has finished
-        self.read_wait.set()
-        self.read_done.wait()
-        self.read_done.clear()
+        return reader
 
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", MagicMock(return_value=True))
-class TestFilteredHLSStream(unittest.TestCase):
-    url_playlist = "http://mocked/test-hls-filtered/playlist.m3u8"
-    url_segment = "http://mocked/test-hls-filtered/stream{0}.ts"
-
-    @classmethod
-    def get_segments(cls, num):
-        return ["[{0}]".format(i).encode("ascii") for i in range(num)]
-
-    @classmethod
-    def get_playlist(cls, media_sequence, items, filtered=False, end=False):
-        playlist = dedent("""
-            #EXTM3U
-            #EXT-X-VERSION:5
-            #EXT-X-TARGETDURATION:1
-            #EXT-X-MEDIA-SEQUENCE:{0}
-        """.format(media_sequence))
-
-        for item in items:
-            playlist += "#EXTINF:1.000,{1}\nstream{0}.ts\n".format(item, "filtered" if filtered else "live")
-
-        if end:
-            playlist += "#EXT-X-ENDLIST\n"
-
-        return playlist
+class TestFilteredHLSStream(TestMixinStreamHLS, unittest.TestCase):
+    __stream__ = _TestSubjectFilteredHLSStream
 
     @classmethod
     def filter_sequence(cls, sequence):
-        return sequence.segment.title == "filtered"
+        return sequence.segment.title == FILTERED
 
-    def setUp(self):
-        self.mock = requests_mock.Mocker()
-        self.mock.start()
-        self.mocks = {}
-
-    def tearDown(self):
-        self.thread.reader.close()
-
-        # make sure that worker, write and read threads halt
+    def close_thread(self):
         self.thread.reader.writer.write_wait.set()
-        self.thread.read_wait.set()
-        self.thread.reader.writer.join()
-        self.thread.reader.worker.join()
-        self.thread.join()
+        super(TestFilteredHLSStream, self).close_thread()
 
-        self.mocks.clear()
-        self.mock.stop()
+    # make one write call on the write thread and wait until it has finished
+    def await_write(self):
+        writer = self.thread.reader.writer
+        writer.write_wait.set()
+        writer.write_done.wait()
+        writer.write_done.clear()
+
+    def get_session(self, options=None, *args, **kwargs):
+        session = super(TestFilteredHLSStream, self).get_session(options)
+        session.set_option("hls-live-edge", 2)
+        session.set_option("hls-timeout", 0)
+        session.set_option("stream-timeout", 0)
 
-    def subject(self, segments, playlists):
-        self.mocks[self.url_playlist] = self.mock.get(self.url_playlist, [{"text": p} for p in playlists])
-        for i, segment in enumerate(segments):
-            url = self.url_segment.format(i)
-            self.mocks[url] = self.mock.get(url, content=segment)
+        return session
 
-        self.thread = _TestSubjectReadThread()
-        self.thread.start()
+    def subject(self, *args, **kwargs):
+        thread, segments = super(TestFilteredHLSStream, self).subject(*args, **kwargs)
 
-        return self.thread, self.thread.reader, self.thread.reader.writer
+        return thread, thread.reader, thread.reader.writer, segments
 
     # don't patch should_filter_sequence here (it always returns False)
     def test_not_filtered(self):
-        segments = self.get_segments(2)
-        thread, reader, writer = self.subject(segments, [
-            self.get_playlist(0, [0, 1], filtered=True, end=True)
+        thread, reader, writer, segments = self.subject([
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)], end=True)
         ])
 
-        thread.await_write()
-        thread.await_write()
-        thread.await_read()
-        self.assertEqual(b"".join(thread.data), b"".join(segments[0:2]), "Does not filter by default")
+        self.await_write()
+        self.await_write()
+        data = self.await_read()
+        self.assertEqual(data, self.content(segments), "Does not filter by default")
 
     @patch("streamlink.stream.hls_filtered.FilteredHLSStreamWriter.should_filter_sequence", new=filter_sequence)
     @patch("streamlink.stream.hls_filtered.log")
     def test_filtered_logging(self, mock_log):
-        segments = self.get_segments(8)
-        thread, reader, writer = self.subject(segments, [
-            self.get_playlist(0, [0, 1], filtered=True),
-            self.get_playlist(2, [2, 3], filtered=False),
-            self.get_playlist(4, [4, 5], filtered=True),
-            self.get_playlist(6, [6, 7], filtered=False, end=True)
+        thread, reader, writer, segments = self.subject([
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)]),
+            Playlist(2, [Segment(2), Segment(3)]),
+            Playlist(4, [SegmentFiltered(4), SegmentFiltered(5)]),
+            Playlist(6, [Segment(6), Segment(7)], end=True)
         ])
+        data = b""
 
         self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
 
         for i in range(2):
-            thread.await_write()
-            thread.await_write()
+            self.await_write()
+            self.await_write()
             self.assertEqual(len(mock_log.info.mock_calls), i * 2 + 1)
             self.assertEqual(mock_log.info.mock_calls[i * 2 + 0], call("Filtering out segments and pausing stream output"))
             self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
 
-            thread.await_write()
-            thread.await_write()
+            self.await_write()
+            self.await_write()
             self.assertEqual(len(mock_log.info.mock_calls), i * 2 + 2)
             self.assertEqual(mock_log.info.mock_calls[i * 2 + 1], call("Resuming stream output"))
             self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
 
-            thread.await_read()
+            data += self.await_read()
 
         self.assertEqual(
-            b"".join(thread.data),
-            b"".join(list(itertools.chain(segments[2:4], segments[6:8]))),
+            data,
+            self.content(segments, cond=lambda s: s.num % 4 > 1),
             "Correctly filters out segments"
         )
-        for i, _ in enumerate(segments):
-            self.assertTrue(self.mocks[self.url_segment.format(i)].called, "Downloads all segments")
+        self.assertTrue(all([self.called(s) for s in segments.values()]), "Downloads all segments")
 
     @patch("streamlink.stream.hls_filtered.FilteredHLSStreamWriter.should_filter_sequence", new=filter_sequence)
     def test_filtered_timeout(self):
-        segments = self.get_segments(2)
-        thread, reader, writer = self.subject(segments, [
-            self.get_playlist(0, [0, 1], filtered=False, end=True)
+        thread, reader, writer, segments = self.subject([
+            Playlist(0, [Segment(0), Segment(1)], end=True)
         ])
 
-        thread.await_write()
-        thread.await_read()
-        self.assertEqual(thread.data, segments[0:1], "Has read the first segment")
+        self.await_write()
+        data = self.await_read()
+        self.assertEqual(data, segments[0].content, "Has read the first segment")
 
         # simulate a timeout by having an empty buffer
         # timeout value is set to 0
-        thread.await_read()
-        self.assertIsInstance(thread.error, IOError, "Raises a timeout error when no data is available to read")
+        with self.assertRaises(IOError) as cm:
+            self.await_read()
+        self.assertEqual(str(cm.exception), "Read timeout", "Raises a timeout error when no data is available to read")
 
     @patch("streamlink.stream.hls_filtered.FilteredHLSStreamWriter.should_filter_sequence", new=filter_sequence)
     def test_filtered_no_timeout(self):
-        segments = self.get_segments(4)
-        thread, reader, writer = self.subject(segments, [
-            self.get_playlist(0, [0, 1], filtered=True),
-            self.get_playlist(2, [2, 3], filtered=False, end=True)
+        thread, reader, writer, segments = self.subject([
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)]),
+            Playlist(2, [Segment(2), Segment(3)], end=True)
         ])
 
         self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
 
-        thread.await_write()
-        thread.await_write()
+        self.await_write()
+        self.await_write()
         self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
 
         # make reader read (no data available yet)
         thread.read_wait.set()
         # once data becomes available, the reader continues reading
-        thread.await_write()
+        self.await_write()
         self.assertTrue(reader.filter_event.is_set(), "Reader is not waiting anymore")
 
         thread.read_done.wait()
         thread.read_done.clear()
         self.assertFalse(thread.error, "Doesn't time out when filtering")
-        self.assertEqual(thread.data, segments[2:3], "Reads next available buffer data")
+        self.assertEqual(b"".join(thread.data), segments[2].content, "Reads next available buffer data")
 
-        thread.await_write()
-        thread.await_read()
-        self.assertEqual(thread.data, segments[2:4])
+        self.await_write()
+        data = self.await_read()
+        self.assertEqual(data, self.content(segments, cond=lambda s: s.num >= 2))
 
     @patch("streamlink.stream.hls_filtered.FilteredHLSStreamWriter.should_filter_sequence", new=filter_sequence)
     def test_filtered_closed(self):
-        segments = self.get_segments(2)
-        thread, reader, writer = self.subject(segments, [
-            self.get_playlist(0, [0, 1], filtered=True)
+        thread, reader, writer, segments = self.subject([
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)])
         ])
 
+        # mock the reader thread's filter_event.wait method, so that the main thread can wait on its call
+        filter_event_wait_called = Event()
+        orig_wait = reader.filter_event.wait
+
+        def mocked_wait(*args, **kwargs):
+            filter_event_wait_called.set()
+            return orig_wait(*args, **kwargs)
+
+        mock = patch.object(reader.filter_event, "wait", side_effect=mocked_wait)
+        mock.start()
+
+        # write first filtered segment and trigger the filter_event's lock
         self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
-        thread.await_write()
+        self.await_write()
         self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
 
         # make reader read (no data available yet)
         thread.read_wait.set()
+        # before calling reader.close(), wait until reader thread's filter_event.wait was called
+        filter_event_wait_called.wait()
 
         # close stream while reader is waiting for filtering to end
-        # FIXME: sleep for 100ms here, so that the reader thread runs into its filter_event before calling close()
-        sleep(0.1)
         thread.reader.close()
         thread.read_done.wait()
         thread.read_done.clear()
         self.assertEqual(thread.data, [b""], "Stops reading on stream close")
         self.assertFalse(thread.error, "Is not a read timeout on stream close")
+
+        mock.stop()