"""HdHub: a second source that hands over plain FILES, not a player.
2026-09-28, MobLand and Tony would not play. Nothing of ours was broken:
Cloudflare had blocked vidcore's CDN host `moon.quietridge.top` for "Terms of
Service violations" -- the whole host, even `/` -- so every master 403'd, from
any referer, and vidcore's OWN player sat on "LOADING THINGS UP...". With vidcore
the only live source, that was every title not already in the library.
cinewave (watch.cinewave.qzz.io) lists HdHub among 22 servers, and it is the one
that is not an embed: a real Stremio addon returning direct mkv links on
Cloudflare R2 and pixeldrain. Measured the same evening from stortron:
* by IMDb id it answers in 0.2-0.4s (by `tmdb:` it answers "No streams found",
after up to 21s) -- MobLand S2E1 27 rows, Tony 33 rows;
* R2 and pixeldrain rows return 206 to a plain Range GET and ffprobe reads
them, at ~10 MB/s -- so ffmpeg can copy them, which vidlink could never do;
* "[10Gbps Download Only]" rows (hubcloud) are HTML/400 -- useless;
* results are fuzzy: Tony (2026) came back with "NCIS: Tony & Ziva" rows;
* many rows are HEVC Main10 or 4K, which the Pi's H.264 decoder cannot play;
* dual-audio files flag HINDI as the default track;
* R2 links are pre-signed for 8h from when HdHub cached them, not from when
we ask -- one handed over at 22:50 UTC expired at 23:25.
Every rule below is one of those facts.
Run: .venv/bin/python3 -m pytest test_hdhub.py -q
"""
import os
import time
import pytest
import addon
# --- fixtures ---------------------------------------------------------------
R2 = ("https://abc.r2.cloudflarestorage.com/hub/x?X-Amz-Date={date}"
"&X-Amz-Expires=28800&X-Amz-Signature=s")
PD = "https://pixeldrain.dev/api/file/{fid}?download"
def _amz_now(offset=0):
return time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(time.time() + offset))
def row(filename, name="HdHub 1080p", host="FSL", size=1_000_000_000, url=None):
if url is None:
url = (R2.format(date=_amz_now()) if host == "FSL"
else PD.format(fid=abs(hash(filename)) % 10 ** 8))
return {"name": name,
"description": f"[{host}] [💾 1 GB] {filename}\nHindi\n{host} | HdHub",
"url": url, "behaviorHints": {"videoSize": size}}
@pytest.fixture(autouse=True)
def isolated(monkeypatch, tmp_path):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "verified.json"))
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_hdhub_cache", {})
monkeypatch.setattr(addon, "_hdhub_probe_cache", {})
monkeypatch.setattr(addon, "_hdhub_pick_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
yield
@pytest.fixture
def tmdb(monkeypatch):
"""Stand in for TMDB. Tests must never reach the network."""
responses = {
"tv/247718": {"name": "MobLand", "first_air_date": "2025-03-30"},
"tv/247718/season/2/episode/1": {"name": "I Wanna Be Your Dog",
"runtime": 55},
"tv/247718/external_ids": {"imdb_id": "tt31510819"},
"movie/1329016": {"title": "Tony", "release_date": "2026-05-01",
"runtime": 100},
"movie/1329016/external_ids": {"imdb_id": "tt33095251"},
}
def fake(path, params=None):
if path not in responses:
raise RuntimeError(f"unexpected TMDB call: {path}")
return responses[path]
monkeypatch.setattr(addon, "_tmdb_get", fake)
return responses
@pytest.fixture
def hdhub(monkeypatch):
"""Stand in for the HdHub addon: {(kind, imdb, s, e): [rows]}."""
listing, calls = {}, []
def fake(kind, imdb, season=0, episode=0):
calls.append((kind, imdb, season, episode))
return listing.get((kind, imdb, season, episode), [])
monkeypatch.setattr(addon, "_hdhub_query", fake)
return type("H", (), {"listing": listing, "calls": calls})()
def probe(codec="h264", height=1080, audio=(("aac", "eng"),), duration=3300.0,
png=False):
streams = [{"index": 0, "codec_type": "video", "codec_name": codec,
"height": height, "disposition": {"attached_pic": 0}}]
for i, (c, lang) in enumerate(audio, start=1):
streams.append({"index": i, "codec_type": "audio", "codec_name": c,
"tags": {"language": lang} if lang else {},
"disposition": {"attached_pic": 0}})
if png:
streams.append({"index": len(streams), "codec_type": "video",
"codec_name": "png", "disposition": {"attached_pic": 1}})
return {"streams": streams, "format": {"duration": str(duration)}}
# --- identity: the right title, or nothing ----------------------------------
@pytest.mark.parametrize("filename", [
"MobLand.S02E01.I.Wanna.Be.Your.Dog.1080p.ATV.WEB-DL.x264.mkv",
"MobLand.2025.S02E01.1080p.WEB-DL.x264.mkv",
"MobLand S02E01 1080p WEB-DL x264.mkv",
])
def test_a_series_file_is_accepted_on_title_and_exact_episode(filename):
assert addon._hdhub_title_ok(filename, "MobLand", "2025", "series", 2, 1)
@pytest.mark.parametrize("filename", [
"MobLand.S02E02.Song.2.1080p.x264.mkv", # wrong episode
"MobLand.S01E01.Stick.or.Twist.1080p.x264.mkv", # wrong season
"MobLand.S02.Complete.1080p.x264.mkv", # a pack, not the episode
"Not.MobLand.S02E01.1080p.x264.mkv", # title must LEAD
])
def test_a_series_file_for_another_episode_or_show_is_refused(filename):
assert not addon._hdhub_title_ok(filename, "MobLand", "2025", "series", 2, 1)
def test_ncis_tony_and_ziva_is_not_tony_2026():
"""The actual rows HdHub returned for Tony (2026)."""
for f in ("Ncis.Tony.&.Ziva.S01E07.1080P.Hindi.English-Vegamovies.is.mkv",
"NCIS Tony and Ziva S01 1080p WEB-DL H264",
"Tony.and.Ziva.2025.1080p.x264.mkv"):
assert not addon._hdhub_title_ok(f, "Tony", "2026", "movie", 0, 0), f
def test_the_real_tony_is_accepted():
f = "Tony.2026.1080p.AMZN.WEB-DL.English.DDP.5.1.Atmos.x264-4kHDHub.com.mkv"
assert addon._hdhub_title_ok(f, "Tony", "2026", "movie", 0, 0)
def test_a_movie_year_may_be_off_by_one_but_not_more():
"""Festival vs wide release straddle a new year; a remake is decades off."""
assert addon._hdhub_title_ok("Tony.2025.1080p.x264.mkv", "Tony", "2026", "movie")
assert not addon._hdhub_title_ok("Tony.2019.1080p.x264.mkv", "Tony", "2026", "movie")
def test_a_movie_request_refuses_an_episode_file():
assert not addon._hdhub_title_ok("Tony.2026.S01E01.1080p.mkv", "Tony", "2026", "movie")
def test_punctuation_and_diacritics_do_not_break_matching():
assert addon._hdhub_title_ok("X-Men.97.S02E08.1080p.x264.mkv",
"X-Men '97", "2024", "series", 2, 8)
assert addon._hdhub_title_ok("Marama.2025.1080p.x264.mkv",
"Mārama", "2025", "movie")
assert addon._hdhub_title_ok("Star.Trek.Strange.New.Worlds.S04E01.1080p.mkv",
"Star Trek: Strange New Worlds", "2022",
"series", 4, 1)
# --- what the Pi can decode -------------------------------------------------
@pytest.mark.parametrize("name,filename", [
("HdHub 1080p", "MobLand.S02E01.1080p.10Bit.WEB-DL.HEVC.x265.mkv"),
("HdHub 1080p", "MobLand.S02E01.1080p.WEB-DL.H.265.mkv"),
("HdHub 2160p", "MobLand.S02E01.2160p.WEB-DL.x264.mkv"),
("4KHDHub 4K", "MobLand.S02E01.2160p.ATV.WEB-DL.mkv"),
("HdHub 1080p", "Film.2020.1080p.BluRay.DTS-HD.MA.x264.mkv"),
("HdHub 1080p", "Film.2020.1080p.BluRay.TrueHD.x264.mkv"),
("HdHub 1080p", "Film.2020.1080p.AV1.mkv"),
])
def test_what_the_pi_cannot_decode_is_refused_by_name(name, filename):
assert not addon._hdhub_decodable(name, filename)
def test_the_4khdhub_brand_is_not_mistaken_for_a_4k_file():
"""'4KHDHub 1080p' is a 1080p row from a site whose name starts with 4K."""
assert addon._hdhub_height("4KHDHub 1080p") == 1080
assert addon._hdhub_decodable("4KHDHub 1080p",
"Tony.2026.1080p.AMZN.WEB-DL.DDP.Atmos.x264-4kHDHub.com.mkv")
# --- hosts ------------------------------------------------------------------
def test_only_hosts_measured_to_serve_bytes_are_used(tmdb, hdhub):
f = "Tony.2026.1080p.AMZN.WEB-DL.English.x264.mkv"
hdhub.listing[("movie", "tt33095251", 0, 0)] = [
row(f, host="10Gbps Download Only", url="https://pixel.hubcloud.ist/x"),
row(f, host="HubDrive", url="https://hubdrive.pics/file/1"),
row(f, host="PixelDrain"),
row(f, host="FSL"),
]
files = addon._hdhub_files("movie", "tmdb:1329016")
hosts = sorted(x.host for x in files)
assert hosts == ["abc.r2.cloudflarestorage.com", "pixeldrain.dev"]
def test_r2_is_preferred_over_pixeldrain_for_the_same_file(tmdb, hdhub):
"""pixeldrain rate-limits hotlinked downloads; R2 does not."""
f = "Tony.2026.1080p.AMZN.WEB-DL.English.x264.mkv"
hdhub.listing[("movie", "tt33095251", 0, 0)] = [row(f, host="PixelDrain"),
row(f, host="FSL")]
assert addon._hdhub_files("movie", "tmdb:1329016")[0].host.endswith("r2.cloudflarestorage.com")
# --- expiry -----------------------------------------------------------------
def test_the_expiry_of_a_presigned_r2_url_is_read_from_the_url():
url = R2.format(date=_amz_now(-8 * 3600 + 600)) # signed 7h50m ago
left = addon._url_seconds_left(url)
assert 500 < left < 700
def test_a_url_without_a_signature_has_no_known_expiry():
assert addon._url_seconds_left(PD.format(fid="abc")) is None
def test_a_nearly_expired_link_is_not_offered(tmdb, hdhub):
f = "Tony.2026.1080p.AMZN.WEB-DL.English.x264.mkv"
dying = R2.format(date=_amz_now(-8 * 3600 + 120)) # 2 minutes left
hdhub.listing[("movie", "tt33095251", 0, 0)] = [row(f, url=dying)]
assert addon._hdhub_files("movie", "tmdb:1329016") == []
def test_ingest_demands_a_wider_margin_than_playback(tmdb, hdhub):
"""A 3.7 GB copy at ~10 MB/s takes ~6 minutes, and a reconnect after expiry
is a 403. 30 minutes left is fine to press play on, not to start a copy."""
f = "Tony.2026.1080p.AMZN.WEB-DL.English.x264.mkv"
url = R2.format(date=_amz_now(-8 * 3600 + 1800))
hdhub.listing[("movie", "tt33095251", 0, 0)] = [row(f, url=url)]
assert addon._hdhub_files("movie", "tmdb:1329016",
margin=addon.HDHUB_PLAY_MARGIN)
assert not addon._hdhub_files("movie", "tmdb:1329016",
margin=addon.HDHUB_INGEST_MARGIN)
# --- language ---------------------------------------------------------------
def test_a_file_with_no_english_audio_is_refused(tmdb, hdhub):
hdhub.listing[("series", "tt31510819", 2, 1)] = [
row("MobLand.S02E01.1080p.WEB-DL.Hindi.AAC.x264.mkv")]
assert addon._hdhub_files("series", "tmdb:247718", 2, 1) == []
def test_english_only_files_rank_above_dual_audio(tmdb, hdhub):
dual = "MobLand.S02E01.1080p.WEB-DL.Hindi.2.0-English.5.1.x264.mkv"
eng = "MobLand.S02E01.1080p.WEB-DL.English.DDP5.1.x264.mkv"
hdhub.listing[("series", "tt31510819", 2, 1)] = [row(dual, size=9e9),
row(eng, size=1e9)]
assert [x.filename for x in addon._hdhub_files("series", "tmdb:247718", 2, 1)] == [eng, dual]
def test_higher_resolution_ranks_first_but_never_above_the_ask(tmdb, hdhub):
hd = "MobLand.S02E01.1080p.WEB-DL.English.x264.mkv"
sd = "MobLand.S02E01.720p.WEB-DL.English.x264.mkv"
hdhub.listing[("series", "tt31510819", 2, 1)] = [row(sd, "HdHub 720p"),
row(hd, "HdHub 1080p")]
assert [x.height for x in addon._hdhub_files("series", "tmdb:247718", 2, 1)] == [1080, 720]
assert [x.height for x in addon._hdhub_files("series", "tmdb:247718", 2, 1,
max_height=720)] == [720]
# --- lookup -----------------------------------------------------------------
def test_hdhub_is_asked_by_imdb_id_even_for_a_tmdb_title(tmdb, hdhub):
"""By `tmdb:` HdHub answers 'No streams found'; by IMDb id it has them."""
addon._hdhub_files("series", "tmdb:247718", 2, 1)
assert hdhub.calls == [("series", "tt31510819", 2, 1)]
def test_an_imdb_title_is_passed_straight_through(tmdb, hdhub, monkeypatch):
monkeypatch.setattr(addon, "_library_meta",
lambda *a, **k: {"tmdb_id": "1329016", "title": "Tony",
"year": "2026", "episode_title": "",
"runtime": 100})
addon._hdhub_files("movie", "tt33095251")
assert hdhub.calls == [("movie", "tt33095251", 0, 0)]
def test_no_imdb_id_means_hdhub_is_not_asked(tmdb, hdhub):
tmdb["movie/1329016/external_ids"] = {"imdb_id": None}
assert addon._hdhub_files("movie", "tmdb:1329016") == []
assert hdhub.calls == []
def test_an_hdhub_outage_yields_nothing_rather_than_an_error(tmdb, monkeypatch):
def boom(*a, **k):
raise RuntimeError("503")
monkeypatch.setattr(addon, "_hdhub_query", boom)
assert addon._hdhub_files("movie", "tmdb:1329016") == []
# --- the probe: the file itself decides -------------------------------------
def test_an_h264_file_with_english_audio_is_usable():
assert addon._probe_usable(probe())
def test_hevc_is_refused_even_when_the_name_did_not_say_so():
assert not addon._probe_usable(probe(codec="hevc"))
def test_a_file_whose_audio_is_all_tagged_hindi_is_refused():
assert not addon._probe_usable(probe(audio=(("aac", "hin"),)))
def test_untagged_audio_is_given_the_benefit_of_the_doubt():
assert addon._probe_usable(probe(audio=(("aac", None),)))
def test_cover_art_is_not_mistaken_for_the_video():
assert addon._probe_usable(probe(png=True))
def test_audio_the_mp4_container_cannot_hold_does_not_count():
assert not addon._probe_usable(probe(audio=(("dts", "eng"),)))
def test_the_pick_skips_a_file_that_fails_its_probe(tmdb, hdhub, monkeypatch):
bad = "MobLand.S02E01.1080p.WEB-DL.English.mkv"
good = "MobLand.S02E01.720p.WEB-DL.English.x264.mkv"
hdhub.listing[("series", "tt31510819", 2, 1)] = [row(bad, "HdHub 1080p"),
row(good, "HdHub 720p")]
monkeypatch.setattr(addon, "_hdhub_probe",
lambda url, f=None: probe(codec="hevc") if "1080" in (f or "") else probe())
picked = addon._hdhub_pick("series", "tmdb:247718", 2, 1, 1080)
assert picked and picked[0].filename == good
def test_no_usable_file_is_none(tmdb, hdhub, monkeypatch):
hdhub.listing[("series", "tt31510819", 2, 1)] = [
row("MobLand.S02E01.1080p.WEB-DL.English.mkv")]
monkeypatch.setattr(addon, "_hdhub_probe", lambda url, f=None: None)
assert addon._hdhub_pick("series", "tmdb:247718", 2, 1, 1080) is None
# --- playback route ---------------------------------------------------------
@pytest.fixture
def client(monkeypatch):
started = []
monkeypatch.setattr(addon, "_ensure_ingest",
lambda *a, **k: started.append(a))
monkeypatch.setattr(addon, "_queue_next_episode", lambda *a, **k: None)
c = addon.app.test_client()
c.started = started
return c
def test_the_route_redirects_to_the_chosen_file(client, monkeypatch):
f = addon.HdHubFile(url="https://x.r2.dev/f.mkv", filename="f", height=1080,
size=1, host="x.r2.dev", english_only=True)
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
monkeypatch.setattr(addon, "_hdhub_pick", lambda *a, **k: (f, probe()))
r = client.get("/hdhub/series/tmdb:247718:2:1.mkv?h=1080")
assert r.status_code == 302 and r.headers["Location"] == "https://x.r2.dev/f.mkv"
assert client.started, "playing is the signal to keep a copy"
def test_a_local_copy_wins_over_the_network(client, monkeypatch, tmp_path):
p = tmp_path / "x.mp4"
p.write_bytes(b"x")
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: str(p))
monkeypatch.setattr(addon, "_hdhub_pick",
lambda *a, **k: pytest.fail("must not look upstream"))
r = client.get("/hdhub/movie/tmdb:1329016.mkv?h=1080")
assert r.status_code == 302 and "/mp4/movie/tmdb:1329016.mp4" in r.headers["Location"]
def test_nothing_usable_is_a_404(client, monkeypatch):
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
monkeypatch.setattr(addon, "_hdhub_pick", lambda *a, **k: None)
assert client.get("/hdhub/movie/tmdb:1329016.mkv").status_code == 404
def test_a_hostile_ident_is_refused(client):
assert client.get("/hdhub/movie/..%2Fetc.mkv").status_code == 404
assert client.get("/hdhub/series/tmdb:1.mkv").status_code == 404
def test_play_and_keep_plays_hdhub_when_there_is_no_copy_yet(client, monkeypatch):
"""The cached 'play & keep' row goes through /mp4/. With vidcore's CDN
banned, a vidcore redirect there is a dead link; a progressive file is the
better live answer anyway -- it seeks."""
f = addon.HdHubFile(url="https://x.r2.dev/f.mkv", filename="f", height=1080,
size=1, host="x.r2.dev", english_only=True)
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
monkeypatch.setattr(addon, "_hdhub_pick", lambda *a, **k: (f, probe()))
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: pytest.fail("hdhub answered first"))
r = client.get("/mp4/movie/tmdb:1329016.mp4?h=1080")
assert r.status_code == 302 and r.headers["Location"] == "https://x.r2.dev/f.mkv"
# --- rows -------------------------------------------------------------------
def test_an_episode_row_carries_the_episode_tag_and_sits_above_vidcore(
client, tmdb, hdhub, monkeypatch):
"""Jacktook drops a TV row without SxxEyy in its title."""
hdhub.listing[("series", "tt31510819", 2, 1)] = [
row("MobLand.S02E01.1080p.WEB-DL.English.x264.mkv")]
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_hdhub_prefetch", lambda *a, **k: None)
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
rows = client.get("/stream/series/fxl:247718:2:1.json").get_json()["streams"]
names = [r["name"] for r in rows]
assert "HdHub" in names
hd = rows[names.index("HdHub")]
assert hd["title"].startswith("S02E01 ")
assert "/hdhub/series/tmdb:247718:2:1.mkv" in hd["url"]
assert names.index("HdHub") < names.index("vidcore")
def test_no_hdhub_files_means_no_hdhub_row(client, tmdb, hdhub, monkeypatch):
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
rows = client.get("/stream/movie/fxl:1329016.json").get_json()["streams"]
assert "HdHub" not in [r["name"] for r in rows]
# --- library ingest ---------------------------------------------------------
@pytest.fixture
def ingest(tmdb, hdhub, monkeypatch):
"""vidcore resolves to a banned CDN; HdHub has the file."""
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_fetch_subtitles_for", lambda *a, **k: None)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://banned/m.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)
def banned(url, dest):
raise RuntimeError("ffmpeg exit 8: Server returned 403 Forbidden")
monkeypatch.setattr(addon, "_remux_to_mp4", banned)
hdhub.listing[("movie", "tt33095251", 0, 0)] = [
row("Tony.2026.1080p.AMZN.WEB-DL.English.x264.mkv")]
monkeypatch.setattr(addon, "_hdhub_probe", lambda url, f=None: probe(duration=6000))
copied = []
def fake_copy(url, p, dest):
copied.append((url, dest))
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
monkeypatch.setattr(addon, "_copy_progressive", fake_copy)
return copied
def test_a_banned_vidcore_falls_back_to_hdhub(ingest):
assert addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080) == "done"
assert ingest and ingest[0][1].endswith("Tony (2026).mp4")
def test_an_unresolvable_vidcore_falls_back_too(ingest, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
assert addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080) == "done"
def test_vidcore_is_still_tried_first(ingest, monkeypatch):
"""vidcore's identity is ASSERTED by TMDB id; HdHub's is matched from a
filename. When both can, the asserted one feeds the library."""
done = []
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: done.append(url)
or open(dest, "wb").write(b"x"))
os.makedirs(os.path.dirname(addon._library_path(
"movie", "Tony", "2026", "1329016")), exist_ok=True)
assert addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080) == "done"
assert done and ingest == []
@pytest.mark.parametrize("seconds", [3000, 9000]) # 50% and 150% of 100 min
def test_a_file_the_wrong_length_for_the_title_is_not_ingested(ingest, monkeypatch, seconds):
"""The last guard on a filename match: a different film or episode is
almost never within 15% of this one's runtime."""
monkeypatch.setattr(addon, "_hdhub_probe", lambda url, f=None: probe(duration=seconds))
state = addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080)
assert state.startswith("failed") and ingest == []
def test_when_both_fail_the_vidcore_reason_is_kept(ingest, hdhub):
hdhub.listing.clear()
state = addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080)
assert state.startswith("failed: ffmpeg exit 8") and "hdhub" in state
def test_a_disabled_hdhub_is_not_consulted(ingest, hdhub, monkeypatch):
monkeypatch.setattr(addon, "SOURCES", {k: v for k, v in addon.SOURCES.items()
if k != "hdhub"})
state = addon._run_ingest("movie", "tmdb:1329016", 0, 0, 1080)
assert state.startswith("failed") and hdhub.calls == []
# --- the copy itself --------------------------------------------------------
def test_the_copy_puts_english_audio_first_and_default(monkeypatch, tmp_path):
"""Kodi plays the DEFAULT track, and these files flag Hindi as default."""
cmds = []
def fake_run(cmd, out, stall_timeout, hard_timeout):
cmds.append(cmd)
open(out, "wb").write(b"x")
return addon._RemuxResult(0, "")
monkeypatch.setattr(addon, "_run_remux", fake_run)
monkeypatch.setattr(addon, "_media_duration", lambda p: 3300.0)
p = probe(audio=(("aac", "hin"), ("eac3", "eng")), png=True)
dest = str(tmp_path / "Show" / "ep.mp4")
addon._copy_progressive("https://x.r2.dev/f.mkv", p, dest)
cmd = cmds[0]
maps = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-map"]
assert maps == ["0:0", "0:2", "0:1"], "video, then English, then the rest"
assert "-disposition:a:0" in cmd and cmd[cmd.index("-disposition:a:0") + 1] == "default"
assert cmd[cmd.index("-disposition:a:1") + 1] == "0"
assert cmd[cmd.index("-f") + 1] == "mp4"
assert os.path.exists(dest) and not os.path.exists(dest + ".part")
def test_a_short_copy_is_not_published(monkeypatch, tmp_path):
"""ffmpeg exits 0 on an input that stops early; measure, as the HLS path does."""
monkeypatch.setattr(addon, "_run_remux", lambda cmd, out, **k: (
open(out, "wb").write(b"x"), addon._RemuxResult(0, ""))[1])
monkeypatch.setattr(addon, "_media_duration", lambda p: 600.0)
dest = str(tmp_path / "ep.mp4")
with pytest.raises(RuntimeError, match="incomplete"):
addon._copy_progressive("https://x.r2.dev/f.mkv", probe(duration=3300), dest)
assert not os.path.exists(dest)
# --- the registry -----------------------------------------------------------
def test_hdhub_is_registered_as_an_ingestable_plain_http_source():
src = addon.SOURCES_DEFAULT["hdhub"]
assert src.enabled and src.ingestable and not src.browser
assert src.container == "progressive-mkv"
def test_vidcore_remains_the_first_choice_for_the_library(monkeypatch):
monkeypatch.setattr(addon, "SOURCES", dict(addon.SOURCES_DEFAULT))
assert [s.id for s in addon.ingest_sources()] == ["vidcore", "hdhub"]
assert addon.ingest_source().id == "vidcore"