OverviewFilesCommitsIssuesDocsReleasesPeers

stremio-vsembed / test_mp4_offline.py

151 lines · python ref:
"""Remuxed MP4 copies: the only way to get seek on this source.

vidcore's HLS is muxed fMP4. Kodi's ffmpeg demuxer plays it (audio works) but
refuses to act on a seek; inputstream.adaptive seeks but cannot play the audio,
because it feeds only the first track of a muxed fMP4. Neither can be fixed from
our side, so instead we stream-copy the episode into a plain progressive MP4,
which ffmpeg seeks natively.

Measured 2026-08-10 on stortron: stream-copy runs at ~5x realtime (120s of video
in 23.9s) and yields h264+aac -- no re-encode, so no quality loss and negligible
CPU. ~1.33GB per 28-minute episode at 1080p, ~0.67GB at 720p.

Run:  .venv/bin/python3 -m pytest test_mp4_offline.py -q
"""

import os

import pytest

import addon


@pytest.fixture(autouse=True)
def tmp_cache(tmp_path, monkeypatch):
    monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path))
    monkeypatch.setattr(addon, "_mp4_jobs", {})
    # _playable_path consults TMDB to derive the library name. These tests cover
    # the LEGACY cache path and must never reach the network -- returning no
    # metadata is exactly the "TMDB unreachable" case, which must still serve a
    # cached copy. Library naming is covered in test_library_ingest.py.
    monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: None)
    yield


# --- naming -----------------------------------------------------------------

def test_series_and_movie_paths_are_distinct_and_stable():
    a = addon._mp4_path("series", "tmdb:138502", 2, 7, 720)
    b = addon._mp4_path("series", "tmdb:138502", 2, 8, 720)
    c = addon._mp4_path("movie", "tmdb:138502", 0, 0, 720)
    assert a != b != c and a != c
    assert a == addon._mp4_path("series", "tmdb:138502", 2, 7, 720)
    assert a.endswith(".mp4")


def test_height_is_part_of_the_name():
    """A 720p copy must not be served to a request for 1080p."""
    assert addon._mp4_path("series", "tmdb:1", 1, 1, 720) != \
        addon._mp4_path("series", "tmdb:1", 1, 1, 1080)


def test_ident_colons_do_not_escape_the_cache_dir():
    """`tmdb:138502` and any hostile ident must stay inside MP4_DIR."""
    for ident in ("tmdb:138502", "../../etc/passwd", "a/b", "tt00/../x"):
        p = os.path.realpath(addon._mp4_path("series", ident, 1, 1, 720))
        assert p.startswith(os.path.realpath(addon.MP4_DIR) + os.sep)


# --- readiness --------------------------------------------------------------

def test_a_partial_download_is_not_ready():
    """A half-written file must never be served: its moov atom is written last,
    so it would be unseekable and truncated."""
    p = addon._mp4_path("series", "tmdb:1", 1, 1, 720)
    open(p + ".part", "wb").write(b"x" * 1024)
    assert addon._mp4_ready("series", "tmdb:1", 1, 1, 720) is False


def test_a_finished_download_is_ready():
    p = addon._mp4_path("series", "tmdb:1", 1, 1, 720)
    open(p, "wb").write(b"x" * 1024)
    assert addon._mp4_ready("series", "tmdb:1", 1, 1, 720) is True


def test_an_empty_file_is_not_ready():
    p = addon._mp4_path("series", "tmdb:1", 1, 1, 720)
    open(p, "wb").close()
    assert addon._mp4_ready("series", "tmdb:1", 1, 1, 720) is False


# --- the route --------------------------------------------------------------

@pytest.fixture
def client(monkeypatch):
    monkeypatch.setattr(addon, "_start_mp4_job", lambda *a, **k: "started")
    return addon.app.test_client()


def test_ready_file_is_served_with_range_support(client):
    p = addon._mp4_path("series", "tmdb:138502", 2, 7, 720)
    with open(p, "wb") as f:
        f.write(b"0123456789" * 100)
    r = client.get("/mp4/series/tmdb:138502:2:7.mp4?h=720")
    assert r.status_code == 200
    assert r.headers.get("Accept-Ranges") == "bytes"

    r = client.get("/mp4/series/tmdb:138502:2:7.mp4?h=720",
                   headers={"Range": "bytes=10-19"})
    assert r.status_code == 206          # seeking depends on this
    assert r.get_data() == b"0123456789"


def test_a_missing_file_falls_through_to_the_live_stream(client, monkeypatch):
    """Was 503. The row has to be playable the instant it is clicked, so a
    missing copy now redirects to the live stream while the copy is made.

    This fixture leaves _library_meta returning None (the TMDB-unreachable
    case), so no ingest is started -- a title we cannot name must not be
    written into the library under a guessed name.
    """
    started = []
    monkeypatch.setattr(addon, "_start_mp4_job",
                        lambda *a, **k: started.append(a) or "started")
    monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/m.m3u8")
    monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)
    r = client.get("/mp4/series/tmdb:138502:2:9.mp4?h=720")
    assert r.status_code == 302
    assert started == [], "unnameable titles are not ingested"


def test_a_bad_ident_is_rejected(client):
    assert client.get("/mp4/series/not-an-id.mp4?h=720").status_code == 404


def test_remux_specifies_the_output_format_explicitly():
    """The output is written to ; without -f mp4 ffmpeg tries to
    infer the muxer from that extension and dies with 'Error initializing the
    muxer ...: Invalid argument'."""
    import inspect
    src = inspect.getsource(addon._remux_to_mp4)
    assert '"-f", "mp4"' in src


def test_a_ready_copy_is_offered_first(monkeypatch):
    monkeypatch.setattr(addon, '_prefetch_vidcore', lambda *a, **k: None)
    open(addon._mp4_path('series', 'tmdb:1', 2, 7, 1080), 'wb').write(b'x')
    with addon.app.test_request_context('/stream/series/fxl:1:2:7.json'):
        out = []
        addon._append_mp4_row(out, 'series', 'tmdb:1', 2, 7)
        addon._append_vidcore_row(out, 'series', 'tmdb:1', 2, 7)
    assert '/mp4/' in out[0]['url'] and 'seekable' in out[0]['title']
    assert len(out) == 1 + len(addon.VIDCORE_HEIGHTS)


def test_no_copy_means_no_row(monkeypatch):
    monkeypatch.setattr(addon, '_prefetch_vidcore', lambda *a, **k: None)
    with addon.app.test_request_context('/stream/series/fxl:1:2:9.json'):
        out = []
        addon._append_mp4_row(out, 'series', 'tmdb:1', 2, 9)
    assert out == []