kodi-jacktook / plugin.video.jacktook / tests / unit / test_adaptive_stream.py

from unittest.mock import MagicMock

from lib.utils.player.utils import adaptive_manifest_type, set_adaptive_stream


# Kodi only offers a video-stream picker in the player OSD when
# inputstream.adaptive is driving playback -- on the bare ffmpeg demuxer it
# picks one variant and never offers a choice. These cover engaging ISA for
# adaptive manifests without disturbing every other kind of source.


def test_detects_hls_and_dash():
    assert adaptive_manifest_type("https://h/x.m3u8") == "hls"
    assert adaptive_manifest_type("https://h/x.mpd") == "mpd"


def test_detection_ignores_the_query_string():
    """Our own stream URLs carry ?h=auto, so a bare-tail match misses them."""
    assert adaptive_manifest_type("http://h:7220/vidcore/series/tmdb:1:2:7.m3u8?h=auto") == "hls"


def test_detection_ignores_the_header_suffix():
    url = "http://h/x.m3u8|Referer=https%3A%2F%2Fvidcore.net%2F"
    assert adaptive_manifest_type(url) == "hls"


def test_non_manifests_are_left_alone():
    for url in ("https://h/video.mp4", "plugin://plugin.video.elementum/play", "magnet:?xt=1", ""):
        assert adaptive_manifest_type(url) is None


def test_engages_isa_for_a_manifest():
    item = MagicMock()
    url = set_adaptive_stream(item, {"adaptive": True, "url": "http://h:7220/a.m3u8?h=auto"})
    assert url == "http://h:7220/a.m3u8?h=auto"
    props = dict(c.args for c in item.setProperty.call_args_list)
    assert props["inputstream"] == "inputstream.adaptive"
    assert props["inputstream.adaptive.manifest_type"] == "hls"


def test_leaves_other_sources_untouched():
    """A torrent/debrid/plugin URL must not be handed to ISA."""
    item = MagicMock()
    assert set_adaptive_stream(item, {"adaptive": True, "url": "https://h/movie.mkv"}) is None
    item.setProperty.assert_not_called()


def test_strips_the_header_suffix_from_the_played_url():
    """Kodi's `|k=v` suffix and ISA's header properties are different
    mechanisms and do not compose; handing the suffixed URL to ISA is how the
    previous attempt fetched the variant playlist bare and took a 403."""
    item = MagicMock()
    url = set_adaptive_stream(
        item, {"adaptive": True, "url": "http://h/a.m3u8|Referer=https%3A%2F%2Fvidcore.net%2F"}
    )
    assert url == "http://h/a.m3u8"
    assert "|" not in url


def test_isa_is_off_unless_a_source_opts_in():
    """Regression, 2026-08-10: engaging ISA for every .m3u8 captured the
    single-variant rows too, and ISA stalls the audio on muxed HLS -- the aac
    decoder opens and then no sound. Content that plays fine on the bare ffmpeg
    demuxer must keep doing so."""
    item = MagicMock()
    assert set_adaptive_stream(item, {"url": "http://h/a.m3u8?h=480"}) is None
    item.setProperty.assert_not_called()


def test_a_url_marker_opts_in():
    """The source marks its own adaptive URL, so no flag has to be plumbed
    through every playback-data path in the fork."""
    item = MagicMock()
    url = set_adaptive_stream(item, {"url": "http://h/a.m3u8?h=auto&adaptive=1"})
    assert url == "http://h/a.m3u8?h=auto&adaptive=1"
    props = dict(c.args for c in item.setProperty.call_args_list)
    assert props["inputstream"] == "inputstream.adaptive"


def test_a_fixed_variant_url_still_does_not_opt_in():
    item = MagicMock()
    assert set_adaptive_stream(item, {"url": "http://h/a.m3u8?h=1080"}) is None
    item.setProperty.assert_not_called()


def test_adaptive_items_are_opened_once():
    """Regression, 2026-08-10: setResolvedUrl AND Player().play() opened the
    same ISA stream twice ~0.7s apart. The second open interrupted the first
    mid-'Finding audio codec', the aac decoder never opened and there was no
    sound, while video and seek worked fine. An adaptive item must be handed
    over once, by setResolvedUrl alone."""
    import os
    src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                            '..', '..', 'lib', 'player.py')).read()
    assert 'double_open = not is_plugin and not getattr(self, "_adaptive", False)' in src
    assert 'if double_open:' in src
    assert 'self._adaptive = bool(adaptive_url)' in src

def test_only_the_two_known_good_properties_are_set():
    """Audio was silent with mimetype and header properties set, and audible
    without them on the same stream. Keep the property set minimal."""
    item = MagicMock()
    set_adaptive_stream(item, {"adaptive": True, "url": "http://h/a.m3u8",
                               "proxyHeaders": {"Referer": "https://vidcore.net/"}})
    props = dict(c.args for c in item.setProperty.call_args_list)
    assert set(props) == {"inputstream", "inputstream.adaptive.manifest_type"}
    item.setMimeType.assert_not_called()